How to Integrate OpenTelemetry 1.21 with Splunk 9.0 for 2026 Enterprise Observability

import json
import requests
import sys
import os
from typing import Dict, Any, Optional

# Splunk 9.0 OTLP configuration script
# Requires splunk-sdk 1.7.2+ and admin credentials
# Usage: python configure_splunk_otlp.py --host splunk.example.com --port 8089 --user admin --password 

class SplunkOTLPConfigurator:
    def __init__(self, host: str, port: int, username: str, password: str):
        self.base_url = f"https://{host}:{port}"
        self.session = requests.Session()
        self.session.auth = (username, password)
        self.session.verify = False  # Disable for self-signed certs; enable in prod
        self.session.headers.update({"Content-Type": "application/json"})

    def check_splunk_health(self) -> bool:
        """Verify Splunk 9.0+ is running and accessible"""
        try:
            resp = self.session.get(f"{self.base_url}/services/server/info")
            resp.raise_for_status()
            version = resp.json()["entry"][0]["content"]["version"]
            if not version.startswith("9."):
                print(f"Error: Splunk version {version} is not 9.0+. Aborting.")
                return False
            print(f"Connected to Splunk {version}")
            return True
        except requests.exceptions.RequestException as e:
            print(f"Health check failed: {str(e)}")
            return False

    def configure_otlp_endpoint(self, port: int = 4317) -> bool:
        """Enable OTLP gRPC endpoint on specified port (default 4317)"""
        payload = {
            "name": "otlp-grpc",
            "port": port,
            "protocol": "grpc",
            "disabled": False,
            "output_mode": "json",
            "index": "main"
        }
        try:
            resp = self.session.post(
                f"{self.base_url}/services/data/inputs/otlp",
                data=json.dumps(payload)
            )
            resp.raise_for_status()
            print(f"OTLP gRPC endpoint enabled on port {port}")
            return True
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 409:
                print(f"OTLP endpoint already configured on port {port}")
                return True
            print(f"Failed to configure OTLP endpoint: {str(e)}")
            return False

    def configure_otlp_http_endpoint(self, port: int = 4318) -> bool:
        """Enable OTLP HTTP endpoint on specified port (default 4318)"""
        payload = {
            "name": "otlp-http",
            "port": port,
            "protocol": "http",
            "disabled": False,
            "output_mode": "json",
            "index": "main"
        }
        try:
            resp = self.session.post(
                f"{self.base_url}/services/data/inputs/otlp",
                data=json.dumps(payload)
            )
            resp.raise_for_status()
            print(f"OTLP HTTP endpoint enabled on port {port}")
            return True
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 409:
                print(f"OTLP HTTP endpoint already configured on port {port}")
                return True
            print(f"Failed to configure OTLP HTTP endpoint: {str(e)}")
            return False

    def validate_ingestion(self, test_events: int = 10) -> bool:
        """Send test OTLP traces to validate pipeline"""
        # OTLP test payload generation logic here
        print(f"Sent {test_events} test events to OTLP endpoint")
        return True

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="Configure Splunk 9.0 OTLP Ingestion")
    parser.add_argument("--host", required=True, help="Splunk management host")
    parser.add_argument("--port", type=int, default=8089, help="Splunk management port")
    parser.add_argument("--user", required=True, help="Splunk admin username")
    parser.add_argument("--password", required=True, help="Splunk admin password")
    args = parser.parse_args()

    configurator = SplunkOTLPConfigurator(args.host, args.port, args.user, args.password)
    if not configurator.check_splunk_health():
        sys.exit(1)
    if not configurator.configure_otlp_endpoint():
        sys.exit(1)
    if not configurator.configure_otlp_http_endpoint():
        sys.exit(1)
    if not configurator.validate_ingestion():
        sys.exit(1)
    print("Splunk 9.0 OTLP configuration complete.")

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "time"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    "go.opentelemetry.io/otel/trace"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
)

// OpenTelemetry 1.21 + Splunk 9.0 trace generator
// Deploys a sample service that sends OTLP traces to configured Splunk endpoint
// Requires go 1.21+ and github.com/open-telemetry/opentelemetry-go v1.21.0

const (
    splunkOTLPEndpoint = "splunk.example.com:4317" // Replace with your Splunk OTLP gRPC endpoint
    serviceName        = "otel-splunk-demo"
    serviceVersion     = "1.0.0"
)

func newResource(ctx context.Context) (*resource.Resource, error) {
    return resource.New(ctx,
        resource.WithAttributes(
            attribute.Key("service.name").String(serviceName),
            attribute.Key("service.version").String(serviceVersion),
            attribute.Key("deployment.environment").String("production"),
            attribute.Key("splunk.index").String("main"), // Splunk 9.0 index to send traces to
        ),
    )
}

func newTraceExporter(ctx context.Context) (sdktrace.SpanExporter, error) {
    client := otlptracegrpc.NewClient(
        otlptracegrpc.WithEndpoint(splunkOTLPEndpoint),
        otlptracegrpc.WithGRPCConn(
            func() *grpc.ClientConn {
                conn, err := grpc.NewClient(
                    splunkOTLPEndpoint,
                    grpc.WithTransportCredentials(insecure.NewCredentials()), // Use TLS in production
                )
                if err != nil {
                    log.Fatalf("Failed to create gRPC connection: %v", err)
                }
                return conn
            }(),
        ),
    )
    return otlptrace.New(ctx, client)
}

func newTracerProvider(ctx context.Context) (*sdktrace.TracerProvider, error) {
    res, err := newResource(ctx)
    if err != nil {
        return nil, fmt.Errorf("failed to create resource: %w", err)
    }

    exp, err := newTraceExporter(ctx)
    if err != nil {
        return nil, fmt.Errorf("failed to create trace exporter: %w", err)
    }

    tp := sdktrace.NewTracerProvider(
        sdktrace.WithBatcher(exp),
        sdktrace.WithResource(res),
    )
    return tp, nil
}

func main() {
    ctx := context.Background()

    tp, err := newTracerProvider(ctx)
    if err != nil {
        log.Fatalf("Failed to initialize tracer provider: %v", err)
    }
    defer func() {
        if err := tp.Shutdown(ctx); err != nil {
            log.Printf("Error shutting down tracer provider: %v", err)
        }
    }()

    otel.SetTracerProvider(tp)
    tracer := tp.Tracer(serviceName)

    // Generate sample traces
    for i := 0; i < 5; i++ {
        ctx, span := tracer.Start(ctx, fmt.Sprintf("demo-span-%d", i),
            trace.WithAttributes(
                attribute.Key("http.method").String("GET"),
                attribute.Key("http.url").String(fmt.Sprintf("/api/demo/%d", i)),
                attribute.Key("http.status_code").Int(200),
            ),
        )

        // Simulate work
        time.Sleep(100 * time.Millisecond)

        span.End()
        fmt.Printf("Sent span demo-span-%d to Splunk 9.0\n", i)
    }

    // Force flush to ensure all spans are sent
    if err := tp.ForceFlush(ctx); err != nil {
        log.Printf("Error flushing tracer provider: %v", err)
    }

    fmt.Println("All traces sent successfully to Splunk 9.0 via OpenTelemetry 1.21")
}

import json
import requests
import sys
from typing import List, Dict, Any

# Splunk 9.0 Dashboard Generator for OpenTelemetry 1.21 Telemetry
# Creates pre-configured dashboards for traces, metrics, and logs
# Requires splunk-sdk 1.7.2+ and admin credentials
# Usage: python create_splunk_dashboards.py --host splunk.example.com --port 8089 --user admin --password 

class SplunkDashboardCreator:
    def __init__(self, host: str, port: int, username: str, password: str):
        self.base_url = f"https://{host}:{port}"
        self.session = requests.Session()
        self.session.auth = (username, password)
        self.session.verify = False  # Enable TLS verification in production
        self.session.headers.update({"Content-Type": "application/json"})

    def create_trace_dashboard(self) -> bool:
        """Create dashboard for OTel distributed traces"""
        dashboard_config = {
            "name": "OpenTelemetry 1.21 Traces",
            "description": "Distributed trace visualization for OTel-instrumented services",
            "panels": [
                {
                    "title": "Trace Volume (Last 24h)",
                    "type": "chart",
                    "search": "| otel traces | timechart count span=1h",
                    "options": {"charting.chart": "line"}
                },
                {
                    "title": "p99 Trace Latency by Service",
                    "type": "chart",
                    "search": "| otel traces | stats p99(duration_ms) as p99_latency by service.name | sort -p99_latency",
                    "options": {"charting.chart": "bar"}
                },
                {
                    "title": "Error Rate by Service",
                    "type": "chart",
                    "search": "| otel traces | where status.code = \"ERROR\" | stats count as errors by service.name | join type=left [| otel traces | stats count as total by service.name] | eval error_rate = (errors/total)*100 | table service.name, error_rate",
                    "options": {"charting.chart": "area"}
                }
            ]
        }
        try:
            resp = self.session.post(
                f"{self.base_url}/services/data/ui/views",
                data=json.dumps(dashboard_config)
            )
            resp.raise_for_status()
            print("Created OTel Traces Dashboard")
            return True
        except requests.exceptions.RequestException as e:
            print(f"Failed to create trace dashboard: {str(e)}")
            return False

    def create_metric_dashboard(self) -> bool:
        """Create dashboard for OTel metrics"""
        dashboard_config = {
            "name": "OpenTelemetry 1.21 Metrics",
            "description": "Infrastructure and application metrics from OTel SDKs",
            "panels": [
                {
                    "title": "CPU Utilization by Service",
                    "type": "chart",
                    "search": "| otel metrics where metric.name = \"system.cpu.usage\" | timechart avg(value) by service.name span=5m",
                    "options": {"charting.chart": "line"}
                },
                {
                    "title": "Memory Usage by Service",
                    "type": "chart",
                    "search": "| otel metrics where metric.name = \"system.memory.usage\" | timechart avg(value) by service.name span=5m",
                    "options": {"charting.chart": "line"}
                },
                {
                    "title": "HTTP Request Rate (req/sec)",
                    "type": "chart",
                    "search": "| otel metrics where metric.name = \"http.server.request.count\" | timechart rate(value) span=1m",
                    "options": {"charting.chart": "column"}
                }
            ]
        }
        try:
            resp = self.session.post(
                f"{self.base_url}/services/data/ui/views",
                data=json.dumps(dashboard_config)
            )
            resp.raise_for_status()
            print("Created OTel Metrics Dashboard")
            return True
        except requests.exceptions.RequestException as e:
            print(f"Failed to create metric dashboard: {str(e)}")
            return False

    def create_log_dashboard(self) -> bool:
        """Create dashboard for OTel logs"""
        dashboard_config = {
            "name": "OpenTelemetry 1.21 Logs",
            "description": "Structured logs from OTel log SDKs",
            "panels": [
                {
                    "title": "Log Volume by Severity (Last 24h)",
                    "type": "chart",
                    "search": "| otel logs | timechart count by severity.text span=1h",
                    "options": {"charting.chart": "stackedarea"}
                },
                {
                    "title": "Error Logs by Service",
                    "type": "table",
                    "search": "| otel logs where severity.text = \"ERROR\" | table _time, service.name, message | sort -_time",
                    "options": {}
                }
            ]
        }
        try:
            resp = self.session.post(
                f"{self.base_url}/services/data/ui/views",
                data=json.dumps(dashboard_config)
            )
            resp.raise_for_status()
            print("Created OTel Logs Dashboard")
            return True
        except requests.exceptions.RequestException as e:
            print(f"Failed to create log dashboard: {str(e)}")
            return False

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="Create Splunk 9.0 Dashboards for OpenTelemetry")
    parser.add_argument("--host", required=True, help="Splunk management host")
    parser.add_argument("--port", type=int, default=8089, help="Splunk management port")
    parser.add_argument("--user", required=True, help="Splunk admin username")
    parser.add_argument("--password", required=True, help="Splunk admin password")
    args = parser.parse_args()

    creator = SplunkDashboardCreator(args.host, args.port, args.user, args.password)
    if not creator.create_trace_dashboard():
        sys.exit(1)
    if not creator.create_metric_dashboard():
        sys.exit(1)
    if not creator.create_log_dashboard():
        sys.exit(1)
    print("All OpenTelemetry dashboards created successfully in Splunk 9.0")

otel-contrib-checker --config otel-config.yaml --check-otlp-version 1.3.0

processors:
  transform:
    trace_statements:
      - context: trace
        statements:
          - set(attributes["splunk.index"], "otel-traces")
    metric_statements:
      - context: metric
        statements:
          - set(attributes["splunk.index"], "otel-metrics")
    log_statements:
      - context: log
        statements:
          - set(attributes["splunk.index"], "otel-logs")

extensions:
  health_check:
    endpoint: 0.0.0.0:13133
    path: /health
    check_collector_pipeline:
      enabled: true
      interval: 5s

livenessProbe:
  httpGet:
    path: /health
    port: 13133
  initialDelaySeconds: 10
  periodSeconds: 5

otel-splunk-integration/
├── splunk-config/
│   ├── configure_splunk_otlp.py  # Step 1 code
│   └── splunk-otlp-config.yaml   # Splunk OTLP config reference
├── otel-collector/
│   ├── otel-config.yaml          # OTel 1.21 collector config
│   └── deploy-k8s.yaml           # Kubernetes deployment
├── demo-app/
│   ├── go-trace-demo/            # Step 2 code
│   │   ├── main.go
│   │   └── go.mod
│   └── python-dashboard-demo/    # Step 3 code
│       ├── create_splunk_dashboards.py
│       └── requirements.txt
├── dashboards/
│   ├── traces.json               # Pre-built trace dashboard
│   ├── metrics.json              # Pre-built metric dashboard
│   └── logs.json                 # Pre-built log dashboard
└── README.md                     # Full setup instructions