Integrate OpenTelemetry with ELK stack for unified observability and distributed tracing

Install Elasticsearch 8

Install Elasticsearch to store OpenTelemetry trace data and metrics. We'll configure it for optimal trace storage performance.

wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo gpg --dearmor -o /usr/share/keyrings/elasticsearch-keyring.gpg
echo "deb [signed-by=/usr/share/keyrings/elasticsearch-keyring.gpg] https://artifacts.elastic.co/packages/8.x/apt stable main" | sudo tee /etc/apt/sources.list.d/elastic-8.x.list
sudo apt update
sudo apt install -y elasticsearch

sudo rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch
cat << 'EOF' | sudo tee /etc/yum.repos.d/elasticsearch.repo
[elasticsearch]
name=Elasticsearch repository for 8.x packages
baseurl=https://artifacts.elastic.co/packages/8.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=0
autorefresh=1
type=rpm-md
EOF
sudo dnf install --enablerepo=elasticsearch -y elasticsearch

Configure Elasticsearch for OpenTelemetry

Configure Elasticsearch with settings optimized for trace data storage and indexing patterns used by OpenTelemetry.

cluster.name: otel-observability
node.name: otel-node-1
path.data: /var/lib/elasticsearch
path.logs: /var/log/elasticsearch
network.host: 0.0.0.0
http.port: 9200
cluster.initial_master_nodes: ["otel-node-1"]
xpack.security.enabled: true
xpack.security.enrollment.enabled: true
xpack.security.http.ssl.enabled: false
xpack.security.transport.ssl.enabled: false
indices.memory.index_buffer_size: 20%
thread_pool.write.queue_size: 1000
bootstrap.memory_lock: true

Configure Elasticsearch memory settings

Set JVM heap size for Elasticsearch to half of available RAM for optimal performance with trace data.

-Xms2g
-Xmx2g

sudo systemctl edit elasticsearch

[Service]
LimitMEMLOCK=infinity

Start Elasticsearch and configure authentication

Start Elasticsearch and set up authentication for secure access from OpenTelemetry components.

sudo systemctl enable --now elasticsearch
sudo systemctl status elasticsearch
sudo /usr/share/elasticsearch/bin/elasticsearch-reset-password -u elastic
sudo /usr/share/elasticsearch/bin/elasticsearch-users useradd otel_writer -p otel_password123 -r superuser

Install Logstash 8

Install Logstash to process and transform OpenTelemetry data before storing it in Elasticsearch.

sudo apt install -y logstash

sudo dnf install --enablerepo=elasticsearch -y logstash

Configure Logstash for OpenTelemetry data processing

Create Logstash configuration to receive OpenTelemetry data via HTTP and process it for Elasticsearch storage.

input {
  http {
    port => 8080
    codec => json
    additional_codecs => {
      "application/x-protobuf" => "plain"
    }
  }
  beats {
    port => 5044
  }
}

filter {
  if [resourceSpans] {
    # Process OpenTelemetry trace data
    ruby {
      code => "
        spans = event.get('resourceSpans')
        if spans
          spans.each do |resource_span|
            resource = resource_span['resource']['attributes'] || {}
            scope_spans = resource_span['scopeSpans'] || []
            
            scope_spans.each do |scope_span|
              spans_array = scope_span['spans'] || []
              
              spans_array.each do |span|
                new_event = LogStash::Event.new()
                new_event.set('[@timestamp]', Time.at(span['startTimeUnixNano'].to_i / 1000000000))
                new_event.set('trace_id', span['traceId'])
                new_event.set('span_id', span['spanId'])
                new_event.set('parent_span_id', span['parentSpanId'])
                new_event.set('operation_name', span['name'])
                new_event.set('duration', span['endTimeUnixNano'].to_i - span['startTimeUnixNano'].to_i)
                new_event.set('service_name', resource['service.name'])
                new_event.set('service_version', resource['service.version'])
                new_event.set('span_kind', span['kind'])
                new_event.set('status_code', span.dig('status', 'code'))
                new_event.set('attributes', span['attributes'])
                new_event.set('events', span['events'])
                new_event.tag('opentelemetry-span')
                
                yield new_event
              end
            end
          end
          event.cancel
        end
      "
    }
  }
  
  if [resourceMetrics] {
    # Process OpenTelemetry metrics data
    ruby {
      code => "
        metrics = event.get('resourceMetrics')
        if metrics
          metrics.each do |resource_metric|
            resource = resource_metric['resource']['attributes'] || {}
            scope_metrics = resource_metric['scopeMetrics'] || []
            
            scope_metrics.each do |scope_metric|
              metrics_array = scope_metric['metrics'] || []
              
              metrics_array.each do |metric|
                new_event = LogStash::Event.new()
                new_event.set('[@timestamp]', Time.now)
                new_event.set('metric_name', metric['name'])
                new_event.set('metric_description', metric['description'])
                new_event.set('metric_unit', metric['unit'])
                new_event.set('service_name', resource['service.name'])
                new_event.set('service_version', resource['service.version'])
                new_event.set('metric_data', metric)
                new_event.tag('opentelemetry-metric')
                
                yield new_event
              end
            end
          end
          event.cancel
        end
      "
    }
  }
  
  # Add common fields
  mutate {
    add_field => { "data_type" => "observability" }
  }
  
  # Parse and enhance trace data
  if "opentelemetry-span" in [tags] {
    mutate {
      add_field => { "index_pattern" => "otel-spans" }
    }
    
    # Convert duration from nanoseconds to milliseconds
    ruby {
      code => "
        duration_ns = event.get('duration')
        if duration_ns
          event.set('duration_ms', duration_ns.to_f / 1000000)
        end
      "
    }
  }
  
  if "opentelemetry-metric" in [tags] {
    mutate {
      add_field => { "index_pattern" => "otel-metrics" }
    }
  }
}

output {
  if "opentelemetry-span" in [tags] {
    elasticsearch {
      hosts => ["localhost:9200"]
      user => "otel_writer"
      password => "otel_password123"
      index => "otel-spans-%{+YYYY.MM.dd}"
      template_name => "otel-spans"
      template_pattern => "otel-spans-*"
      template => {
        "index_patterns" => ["otel-spans-*"]
        "settings" => {
          "number_of_shards" => 1
          "number_of_replicas" => 0
          "refresh_interval" => "5s"
        }
        "mappings" => {
          "properties" => {
            "@timestamp" => { "type" => "date" }
            "trace_id" => { "type" => "keyword" }
            "span_id" => { "type" => "keyword" }
            "parent_span_id" => { "type" => "keyword" }
            "operation_name" => { "type" => "text", "fields" => { "keyword" => { "type" => "keyword" } } }
            "service_name" => { "type" => "keyword" }
            "service_version" => { "type" => "keyword" }
            "duration" => { "type" => "long" }
            "duration_ms" => { "type" => "float" }
            "span_kind" => { "type" => "keyword" }
            "status_code" => { "type" => "keyword" }
          }
        }
      }
    }
  }
  
  if "opentelemetry-metric" in [tags] {
    elasticsearch {
      hosts => ["localhost:9200"]
      user => "otel_writer"
      password => "otel_password123"
      index => "otel-metrics-%{+YYYY.MM.dd}"
      template_name => "otel-metrics"
      template_pattern => "otel-metrics-*"
      template => {
        "index_patterns" => ["otel-metrics-*"]
        "settings" => {
          "number_of_shards" => 1
          "number_of_replicas" => 0
          "refresh_interval" => "5s"
        }
        "mappings" => {
          "properties" => {
            "@timestamp" => { "type" => "date" }
            "metric_name" => { "type" => "keyword" }
            "service_name" => { "type" => "keyword" }
            "service_version" => { "type" => "keyword" }
          }
        }
      }
    }
  }
  
  stdout {
    codec => rubydebug
  }
}

Install Kibana 8

Install Kibana for visualizing OpenTelemetry data and creating observability dashboards.

sudo apt install -y kibana

sudo dnf install --enablerepo=elasticsearch -y kibana

Configure Kibana for observability

Configure Kibana to connect to Elasticsearch and enable features for distributed tracing visualization.

server.port: 5601
server.host: "0.0.0.0"
server.publicBaseUrl: "http://203.0.113.10:5601"
elasticsearch.hosts: ["http://localhost:9200"]
elasticsearch.username: "otel_writer"
elasticsearch.password: "otel_password123"
logging.appenders.file.type: file
logging.appenders.file.fileName: /var/log/kibana/kibana.log
logging.appenders.file.layout.type: json
logging.root.appenders: [default, file]
logging.root.level: info
telemetry.enabled: false
data.search.aggs.shardDelay.enabled: true
xpack.apm.enabled: true
xpack.observability.enabled: true

Install OpenTelemetry Collector

Download and install the OpenTelemetry Collector to receive telemetry data from applications.

wget https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v0.91.0/otelcol-contrib_0.91.0_linux_amd64.tar.gz
tar -xzf otelcol-contrib_0.91.0_linux_amd64.tar.gz
sudo mv otelcol-contrib /usr/local/bin/
sudo chmod +x /usr/local/bin/otelcol-contrib
sudo mkdir -p /etc/otelcol
sudo mkdir -p /var/log/otelcol
sudo useradd --system --no-create-home --shell /bin/false otelcol
sudo chown -R otelcol:otelcol /var/log/otelcol

Configure OpenTelemetry Collector

Create comprehensive collector configuration to receive traces and metrics from applications and export to Logstash.

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318
  jaeger:
    protocols:
      grpc:
        endpoint: 0.0.0.0:14250
      thrift_http:
        endpoint: 0.0.0.0:14268
      thrift_compact:
        endpoint: 0.0.0.0:6831
      thrift_binary:
        endpoint: 0.0.0.0:6832
  zipkin:
    endpoint: 0.0.0.0:9411
  prometheus:
    config:
      scrape_configs:
        - job_name: 'otel-collector'
          scrape_interval: 30s
          static_configs:
            - targets: ['localhost:8888']
  hostmetrics:
    collection_interval: 30s
    scrapers:
      cpu:
      memory:
      load:
      disk:
      filesystem:
      network:
      process:

processors:
  batch:
    send_batch_size: 1024
    send_batch_max_size: 2048
    timeout: 5s
  memory_limiter:
    limit_mib: 512
    spike_limit_mib: 128
    check_interval: 5s
  resource:
    attributes:
      - key: deployment.environment
        value: production
        action: upsert
      - key: collector.version
        value: "0.91.0"
        action: upsert
  attributes:
    actions:
      - key: http.user_agent
        action: delete
      - key: http.request.header.authorization
        action: delete
  span:
    name:
      to_attributes:
        rules:
          - ^/api/v1/(?P.?)/(?P.?)$
      from_attributes: ["http.method", "http.route"]
      separator: " "

exporters:
  logging:
    loglevel: info
  elasticsearch:
    endpoints: ["http://localhost:9200"]
    user: otel_writer
    password: otel_password123
    traces_index: otel-spans
    metrics_index: otel-metrics
    logs_index: otel-logs
    pipeline: otel-pipeline
    timeout: 30s
    retry_on_failure:
      enabled: true
      initial_interval: 5s
      max_interval: 30s
      max_elapsed_time: 300s
  otlphttp/logstash:
    endpoint: "http://localhost:8080"
    timeout: 30s
    retry_on_failure:
      enabled: true
      initial_interval: 5s
      max_interval: 30s
      max_elapsed_time: 300s
    headers:
      "Content-Type": "application/json"
  prometheus:
    endpoint: "0.0.0.0:8889"
    const_labels:
      environment: production
      collector: otelcol-contrib

service:
  telemetry:
    logs:
      level: info
      development: false
      encoding: json
      output_paths: ["/var/log/otelcol/otelcol.log"]
      error_output_paths: ["/var/log/otelcol/otelcol-error.log"]
    metrics:
      address: 0.0.0.0:8888
      level: detailed
  extensions: [health_check, pprof, zpages]
  pipelines:
    traces:
      receivers: [otlp, jaeger, zipkin]
      processors: [memory_limiter, resource, attributes, span, batch]
      exporters: [elasticsearch, logging]
    metrics:
      receivers: [otlp, prometheus, hostmetrics]
      processors: [memory_limiter, resource, batch]
      exporters: [elasticsearch, prometheus, logging]
    logs:
      receivers: [otlp]
      processors: [memory_limiter, resource, batch]
      exporters: [elasticsearch, logging]

extensions:
  health_check:
    endpoint: 0.0.0.0:13133
  pprof:
    endpoint: 0.0.0.0:1777
  zpages:
    endpoint: 0.0.0.0:55679

Create systemd services for OpenTelemetry Collector

Create systemd service files to manage the OpenTelemetry Collector as a system service.

[Unit]
Description=OpenTelemetry Collector
Documentation=https://opentelemetry.io/docs/collector/
After=network.target
Wants=network-online.target

[Service]
Type=exec
User=otelcol
Group=otelcol
ExecStart=/usr/local/bin/otelcol-contrib --config=/etc/otelcol/config.yaml
ExecReload=/bin/kill -HUP $MAINPID
KillMode=mixed
KillSignal=SIGTERM
TimeoutStopSec=30
Restart=on-failure
RestartSec=5
NotifyAccess=none
LimitNOFILE=65536
StandardOutput=journal
StandardError=journal
SyslogIdentifier=otelcol

[Install]
WantedBy=multi-user.target

Start all services

Enable and start all components of the observability stack in the correct order.

sudo systemctl daemon-reload
sudo systemctl enable --now elasticsearch
sudo systemctl enable --now logstash
sudo systemctl enable --now kibana
sudo systemctl enable --now otelcol
sudo systemctl status elasticsearch logstash kibana otelcol

Configure firewall rules

Open necessary ports for OpenTelemetry data ingestion and dashboard access.

sudo ufw allow 5601/tcp comment "Kibana"
sudo ufw allow 4317/tcp comment "OpenTelemetry gRPC"
sudo ufw allow 4318/tcp comment "OpenTelemetry HTTP"
sudo ufw allow 14268/tcp comment "Jaeger HTTP"
sudo ufw allow 9411/tcp comment "Zipkin"
sudo ufw allow 8888/tcp comment "OTel metrics"
sudo ufw allow 8889/tcp comment "Prometheus export"
sudo ufw reload

sudo firewall-cmd --permanent --add-port=5601/tcp
sudo firewall-cmd --permanent --add-port=4317/tcp
sudo firewall-cmd --permanent --add-port=4318/tcp
sudo firewall-cmd --permanent --add-port=14268/tcp
sudo firewall-cmd --permanent --add-port=9411/tcp
sudo firewall-cmd --permanent --add-port=8888/tcp
sudo firewall-cmd --permanent --add-port=8889/tcp
sudo firewall-cmd --reload

Create Kibana index patterns and dashboards

Set up index patterns in Kibana to visualize OpenTelemetry data and create observability dashboards.

curl -X POST "localhost:5601/api/saved_objects/index-pattern/otel-spans" \
  -H "Content-Type: application/json" \
  -H "kbn-xsrf: true" \
  -u "otel_writer:otel_password123" \
  -d '{
    "attributes": {
      "title": "otel-spans-*",
      "timeFieldName": "@timestamp"
    }
  }'

curl -X POST "localhost:5601/api/saved_objects/index-pattern/otel-metrics" \
  -H "Content-Type: application/json" \
  -H "kbn-xsrf: true" \
  -u "otel_writer:otel_password123" \
  -d '{
    "attributes": {
      "title": "otel-metrics-*",
      "timeFieldName": "@timestamp"
    }
  }'

Example Node.js application instrumentation

Configure a sample Node.js application to send traces to your OpenTelemetry Collector.

{
  "name": "otel-demo-app",
  "version": "1.0.0",
  "dependencies": {
    "@opentelemetry/api": "^1.7.0",
    "@opentelemetry/sdk-node": "^0.45.0",
    "@opentelemetry/instrumentation": "^0.45.0",
    "@opentelemetry/exporter-otlp-http": "^0.45.0",
    "@opentelemetry/resources": "^1.18.0",
    "@opentelemetry/semantic-conventions": "^1.18.0",
    "express": "^4.18.0"
  }
}

const { NodeSDK } = require('@opentelemetry/sdk-node');
const { OTLPTraceExporter } = require('@opentelemetry/exporter-otlp-http');
const { Resource } = require('@opentelemetry/resources');
const { SemanticResourceAttributes } = require('@opentelemetry/semantic-conventions');
const express = require('express');

// Initialize OpenTelemetry SDK
const sdk = new NodeSDK({
  resource: new Resource({
    [SemanticResourceAttributes.SERVICE_NAME]: 'demo-service',
    [SemanticResourceAttributes.SERVICE_VERSION]: '1.0.0',
    [SemanticResourceAttributes.DEPLOYMENT_ENVIRONMENT]: 'production',
  }),
  traceExporter: new OTLPTraceExporter({
    url: 'http://localhost:4318/v1/traces',
  }),
});

sdk.start();

const app = express();
const port = 3000;

app.get('/', (req, res) => {
  res.json({ message: 'Hello from instrumented app!', timestamp: new Date().toISOString() });
});

app.get('/api/users/:id', async (req, res) => {
  const userId = req.params.id;
  // Simulate some work
  await new Promise(resolve => setTimeout(resolve, Math.random() * 100));
  res.json({ userId, name: User ${userId}, timestamp: new Date().toISOString() });
});

app.listen(port, () => {
  console.log(Demo app listening at http://localhost:${port});
});

Example Python application instrumentation

Configure a sample Python Flask application with OpenTelemetry instrumentation.

flask==2.3.3
opentelemetry-distro==0.42b0
opentelemetry-exporter-otlp==1.21.0
opentelemetry-instrumentation-flask==0.42b0
opentelemetry-instrumentation-requests==0.42b0

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.semconv.resource import ResourceAttributes
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from flask import Flask, jsonify
import time
import random

Configure OpenTelemetry
resource = Resource.create({
    ResourceAttributes.SERVICE_NAME: "python-demo-service",
    ResourceAttributes.SERVICE_VERSION: "1.0.0",
    ResourceAttributes.DEPLOYMENT_ENVIRONMENT: "production",
})

trace.set_tracer_provider(TracerProvider(resource=resource))
tracer = trace.get_tracer(__name__)

Configure OTLP exporter
otlp_exporter = OTLPSpanExporter(
    endpoint="http://localhost:4318/v1/traces",
    headers={}
)

span_processor = BatchSpanProcessor(otlp_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)

Create Flask app
app = Flask(__name__)
FlaskInstrumentor().instrument_app(app)

@app.route('/')
def hello():
    return jsonify({
        "message": "Hello from Python instrumented app!",
        "timestamp": time.time()
    })

@app.route('/api/process/')
def process_task(task_id):
    with tracer.start_as_current_span("process_task") as span:
        span.set_attribute("task.id", task_id)
        span.set_attribute("task.type", "background_job")
        
        # Simulate processing
        processing_time = random.uniform(0.1, 0.5)
        time.sleep(processing_time)
        
        span.set_attribute("task.duration_seconds", processing_time)
        span.set_attribute("task.status", "completed")
        
        return jsonify({
            "task_id": task_id,
            "status": "completed",
            "processing_time": processing_time
        })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)