> ## Documentation Index
> Fetch the complete documentation index at: https://docs.mezmo.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Log Data Optimization Pipeline
> Build a responsive log pipeline that enriches, parses, routes by state, aggregates, and samples logs before sending them to Mezmo Log Analysis.
***
Why it matters
Log data can be incredibly valuable, but this is often dependent on context and time. By utilizing the Mezmo Data Profiles along with [Responsive Pipelines](/telemetry-pipelines/configure-responsive-pipelines) and proper parsing, teams can ensure the right data shows up at the right time in the right place with the right format.
## Step 1: Create a new Pipeline to handle and route OpenTelemetry Logs
We will begin by routing OpenTelemetry Logs to Mezmo Log Analysis. To do this while optimizing the data, create a new Mezmo Pipeline by clicking [New Pipeline](https://app.mezmo.com/pipelines/pipeline/new) in the platform. Give this a name like `Log Handler`.
## Step 2: Add OpenTelemetry Log Source
Click `Add Source` and select your OpenTelemetry Log source from the `Shared Sources` list just like you did for the `Exploration` pipeline in the prior section.
## Step 3: Insert State Enrichment and Normalize
In order to take advantage of [Responsive Pipelines](/telemetry-pipelines/configure-responsive-pipelines), we need to enrich each event with the current pipelines operational state. The easiest way to do this is by adding an enrichment script right off the bat. Click the `three dots` on your Otel Log Source and select `Add Node->Add Processor->Script Execution`.
Paste in the following JavaScript and click `Save`. Note that the script does a bit more than add the `operational_state` state variable, we are also normalizing some of this Otel data to be easier to use in Mezmo Log Analysis.
```javascript theme={null}
function processEvent(message, metadata, timestamp, annotations) {
metadata.resource.attributes["pipeline.path"] = "with_mezmo"
const state = getPipelineStateVariable("operational_state")
let line = message
let app = metadata.resource.attributes["container.name"]
let host = metadata.resource.attributes["container.hostname"]
let level = metadata.level
if( app == null || app == '' ){
app = metadata.resource["service.name"]
}
if( app == null || app == '' ){
app = metadata.resource["service_name"]
}
if( app == null || app == '' ){
app = metadata.scope.name
}
if( app == null || app == '' ){
app = 'na'
}
if( host == null || host == '' ){
host = metadata.headers["x-bus-partition-key"] ?? metadata.headers["x-kafka-partition-key"]
}
if( host == null || host == '' ){
host = metadata.attributes["log.file.path"]
}
if( host == null || host == '' ){
host = 'na'
}
if( level == null || level == '' ){
level = annotations.level
}
metadata.headers = null
let new_msg = {
"line":line,
"app":app,
"host":host,
"level": level,
"op_state":state,
"meta":metadata,
'_cnt': 1
}
if( message == null ){ return null }
return new_msg
}
```
**Pro Tip**
Note that one can assign many state variables to a single pipeline. This allows for different complex workflows to be automated like QA, curated deployment telemetry and more.
## Step 4: Parse Custom Apache Logs
As noted in the prior section, the raw Apache data coming in from `frontend-proxy` is an unparsed, custom [format defined by the OpenTelemetry demo](https://github.com/braxtonj/opentelemetry-demo/blob/main/src/frontend-proxy/envoy.tmpl.yaml#L80). To ensure teams can take full advantage, we will parse this data with a Grok parser so that it is structured and easily searchable.
To do this, we will add another `Route` Processor by App along with a Parse Sequentially processor on that slice of data.
### Step 4.a: App Router
Connect a Route processor through the `three dots` or `Add Processor` at the bottom with the following configuration on application
* Title: `App Router`
* Route:
* Title: `Frontend Proxy`
* Criteria: `message.app` `equal` `frontend-proxy`
Hit save.
### Step 4.b: Parse Sequentially Processor
Now let's wire a `Parse Sequentially` processor to the `Frontend Proxy` output. We will use the following configuration
* Field: `message.line`
* Target field: `message.line_parsed`
* Custom Parser:
* Title: `Custom Apache`
* Parser: `Grok Pattern`
* Pattern:
`%{SQUARE_BRACKET}%{TIMESTAMP_ISO8601:dt}%{SQUARE_BRACKET} %{DOUBLE_QUOTE}%{DATA:method} %{DATA:path} %{DATA:http_protocol}%{DOUBLE_QUOTE} %{DATA:rsp_code} %{DATA:rsp_flags} %{DATA:rsp_code_details} %{DATA:conn_term_details} %{DOUBLE_QUOTE}%{DATA:upstream_transport_failure_reason}%{DOUBLE_QUOTE} %{DATA:bytes_received} %{DATA:bytes_sent} %{DATA:duration} %{DATA:rsp_upstream_service_time} %{DOUBLE_QUOTE}%{DATA:req_forward_for}%{DOUBLE_QUOTE} %{DOUBLE_QUOTE}%{DATA:req_user_agent}%{DOUBLE_QUOTE} %{DOUBLE_QUOTE}%{DATA:req_id}%{DOUBLE_QUOTE} %{DOUBLE_QUOTE}%{DATA:req_authority}%{DOUBLE_QUOTE} %{DOUBLE_QUOTE}%{DATA:upstream_host}%{DOUBLE_QUOTE} %{DATA:upstream_cluster} %{DATA:upstream_local_addr} %{DATA:downstream_local_addr} %{DATA:downstream_remote_addr} %{DATA:requested_server_name} %{GREEDYDATA:route_name}`
Finally, let's keep the original line with a quick Script, add a `Script Execution` processor connected to the `Parse Sequentially: Apache` output containing the following script:
```javascript theme={null}
function processEvent(message, metadata, timestamp, annotations) {
let old_line = message.line
message.line = message.line_parsed
message.line.message = old_line
message.line_parsed = null
return message
}
```
Your pipeline should look like this at this point.
## Step 5: Route Based on State
After the initial Enrichment processor and Apache Parser, let's now route the data flow based on Pipelines state, i.e. that `operational_state`. Connect a Route processor to all prior outputs with the following configuration
* Title: `State Router`
* Route 1:
* Title: `Normal`
* Criteria: `message.op_state` `contains` `normal`
* Route 2:
* Title: `Incident`
* Criteria: `message.op_state` `contains` `incident`
* Route 3:
* Title: `Deploy`
* Criteria: `message.op_state` `contains` `deploy`
You will end up with a pipeline that looks like the following
## Step 6: Aggregate Normal State Logs
Now, lets optimize those noisy homepage flooding logs using a `Reduce` processor. Effectively this will let us count and aggregate mainly logs into one easy to watch signal. A common pattern is to turn this into a metric but today we will leave this as a log for the downstream tool.
To accomplish this, create a new Route processor connected to `State Router`'s `Normal` and `Unmatched` outputs with the following configuration:
* Title: `Template Router`
* Route 1:
* Title: `Product Found`
* Criteria: `(message.app equal product-catalog) AND (message.line contains Product Found)`
We will then add a `Reduce` processor to roll these logs up over 5 minutes. Connect the processor with the following configuration
* Title: `5min Count`
* Duration: `5 minutes`
* Group By Field Path: `message.host`
* Merge Strategy per Field:
* Field Path: `message._cnt` `sum`
Finally, we will convert the output into a summary message using the following configuration
```javascript theme={null}
function processEvent(message, metadata, timestamp, annotations) {
message.line = {
'message': message._cnt.toString() + ' products found via product-catalog',
'count': message._cnt
}
return message
}
```
At this point, your pipeline should look like the following
## Step 7: Sample Normal State Logs
From `Template Router: Unmatached` let's add sample 1 out of every 10 while in `Normal` state Simply connect a `Sample` processor with the following configuration:
* Rate: `1/10`
## Step 8: Wire to Mezmo Log Analysis
Finally, we will send all of this data into Mezmo Log Analysis. Because of our earlier work normalizing data in Step 3, we can simply add a final Destination to all nodes (including the `Incident` and `Deploy` paths). Create a `Mezmo Log Analysis` destination with the following configuration
* Ingestion key: generate or select
* Query Parameters:
* Hostname: `{{message.host}}`
* Tags: `otel-demo`
* Log Construction Scheme:
* Scheme: `Message pass-through`
## Step 9: Deploy
Next, we must deploy your pipeline in order to start optimizing that log data.
To do this, select `Deploy` in the top right of the UX
## Step 10: Initiate State and Grab State ID
Our final step is to initiate and grab the state ID for the pipeline for automation and control later on. To do this, first we will enact a state by toggling the Normal mode to Incident mode and back.
Now that has been initiated, you will need to grab the `Log Handler` pipeline's ID (found in the URL at `app.mezmo.com/ACCOUNT_ID/pipelines/PIPELINE_ID`) along with a Pipeline API Key here. Then, modify the following script with both that `PIPELINE_ID` and Pipeline API Key
```bash theme={null}
curl --request GET \
--url 'https://api.mezmo.com/v3/pipeline/state-variable?pipeline_id=PIPELINE_ID' \
--header 'Authorization: Token PIPELINE_API_KEY'
```
**Regions outside the US**
Note that for different deployment regions, the url will change slightly. For instance, in the EU one would use `https://api.eu.mezmo.com/v3/pipeline/state-variable?pipeline_id=PIPELINE_ID`
Take the response and save the `STATE_ID` for later. You will find it in the `operational_state`'s data packet, which should look something akin to
```bash theme={null}
{
"meta": {
"pk": "id",
"type": "pipeline-state-variable",
"links": {
"self": {
"create": {
"uri": "/v3/pipeline/{pipeline_id}/state-variable",
"method": "post"
},
"list": {
"uri": "/v3/pipeline/{pipeline_id}/state-variable",
"method": "get"
},
"replace": {
"uri": "/v3/pipeline/{pipeline_id}/state-variable/{id}",
"method": "put"
},
"update": {
"uri": "/v3/pipeline/{pipeline_id}/state-variable",
"method": "patch"
},
"detail": null
},
"related": {
"pipeline": {
"list": "/v3/pipeline",
"detail": "/v3/pipeline/{pipeline_id}"
}
}
},
"page": {
"next": null,
"previous": null
}
},
"data": [
{
"id": "STATE_ID",
"account_id": "ACCOUNT_ID",
"pipeline_id": "PIPELINE_ID",
"state": {
"operational_state": "normal"
},
"created_at": "UTC Timestamp",
"updated_at": "UTC Timestamp"
}
]
}
```
## Step 11: View In Mezmo Log Analysis
Navigate to [Log Analysis](https://app.mezmo.com/logs/view) and view the incoming data. In particular, if you used the `tag` above you can simply search for `tag:otel-demo`.
Let's look at the outcome of our prior Log Aggregation and Parsing.
First, look for the aggregated data by searching for `tag:otel-demo "products found via product-catalog"`. Notice that instead of raw lines like we saw in the Profile, we now have an aggregated message to watch saving tens of thousands of log lines.
Second, check out the newly parsed data by searching for `tag:otel-demo app:frontend-proxy`. While logs are displayed nicely in the Log Viewer, one can expand a line and see all the nested structure that is easily searchable. For instance, to see all 2xx responses enter the query `tag:otel-demo app:frontend-proxy resp_code:(>=200 AND <300)`
**Learn About Log Analysis**
If you want to learn more about Log Analysis and creating things like saved Views, Alerts and more check out our [docs here](/docs) or reaching out to [support@mezmo.com](mailto:support@mezmo.com)
Now let's move on and see explore some simple but powerful optimizations for OpenTelemetry Metrics and Traces.