We Ditched Splunk for Datadog 7.45 and Splunk Cloud: 35% Lower Costs

import json
import logging
import os
import time
from dataclasses import dataclass
from typing import List, Dict, Optional
import requests
from requests.exceptions import RequestException

# Configuration dataclass for migration settings
@dataclass
class MigrationConfig:
    splunk_hec_url: str
    splunk_hec_token: str
    datadog_site: str  # e.g., datadoghq.com, datadoghq.eu
    datadog_api_key: str
    batch_size: int = 1000
    max_retries: int = 3
    retry_delay: int = 5  # seconds

# Setup logging with structured output
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(name)s - %(levelname)s - %(message)s\"
)
logger = logging.getLogger(__name__)

class SplunkToDatadogMigrator:
    def __init__(self, config: MigrationConfig):
        self.config = config
        self.splunk_session = requests.Session()
        self.splunk_session.headers.update({
            \"Authorization\": f\"Splunk {self.config.splunk_hec_token}\",
            \"Content-Type\": \"application/json\"
        })
        self.datadog_api_base = f\"https://http-intake.logs.{self.config.datadog_site}\"
        self.datadog_headers = {
            \"DD-API-KEY\": self.config.datadog_api_key,
            \"Content-Type\": \"application/json\"
        }

    def fetch_splunk_logs(self, start_time: int, end_time: int, index: str = \"*\") -> List[Dict]:
        \"\"\"Fetch logs from Splunk HEC within a time range. Returns empty list on failure.\"\"\"
        endpoint = f\"{self.config.splunk_hec_url}/services/search/jobs\"
        search_query = f\"search index={index} earliest={start_time} latest={end_time} | fields *\"
        payload = {
            \"search\": search_query,
            \"output_mode\": \"json\",
            \"exec_mode\": \"blocking\"
        }

        for attempt in range(self.config.max_retries):
            try:
                response = self.splunk_session.post(endpoint, json=payload, timeout=30)
                response.raise_for_status()
                job_id = response.json().get(\"sid\")
                if not job_id:
                    logger.error(\"No job ID returned from Splunk\")
                    return []

                # Poll for job completion
                results_endpoint = f\"{self.config.splunk_hec_url}/services/search/jobs/{job_id}/results\"
                while True:
                    results_response = self.splunk_session.get(results_endpoint, timeout=30)
                    results_response.raise_for_status()
                    results_data = results_response.json()
                    if results_data.get(\"done\"):
                        return results_data.get(\"results\", [])
                    time.sleep(self.config.retry_delay)
            except RequestException as e:
                logger.warning(f\"Attempt {attempt + 1} failed: {str(e)}\")
                if attempt < self.config.max_retries - 1:
                    time.sleep(self.config.retry_delay * (2 ** attempt))  # Exponential backoff
                else:
                    logger.error(f\"Failed to fetch Splunk logs after {self.config.max_retries} attempts\")
                    return []

    def transform_log_for_datadog(self, splunk_log: Dict) -> Dict:
        \"\"\"Map Splunk fields to Datadog log schema. Strips Splunk internal fields.\"\"\"
        datadog_log = {
            \"message\": splunk_log.get(\"raw\", json.dumps(splunk_log)),
            \"timestamp\": int(splunk_log.get(\"_time\", time.time())),
            \"host\": splunk_log.get(\"host\", \"unknown\"),
            \"service\": splunk_log.get(\"sourcetype\", \"unknown\"),
            \"ddsource\": splunk_log.get(\"source\", \"splunk-migration\"),
            \"tags\": [f\"splunk_index:{splunk_log.get(\"index\", \"unknown\")}\"]
        }
        # Add custom fields except Splunk internals
        excluded_fields = {\"_time\", \"raw\", \"host\", \"sourcetype\", \"source\", \"index\", \"_splunk_*}\"
        for key, value in splunk_log.items():
            if key not in excluded_fields and not key.startswith(\"_splunk_\"):
                datadog_log[\"tags\"].append(f\"{key}:{value}\")
        return datadog_log

    def send_to_datadog(self, logs: List[Dict]) -> bool:
        \"\"\"Send batch of logs to Datadog. Returns True on success.\"\"\"
        endpoint = f\"{self.datadog_api_base}/v1/input\"
        for attempt in range(self.config.max_retries):
            try:
                response = requests.post(
                    endpoint,
                    headers=self.datadog_headers,
                    json=logs,
                    timeout=30
                )
                response.raise_for_status()
                logger.info(f\"Successfully sent {len(logs)} logs to Datadog\")
                return True
            except RequestException as e:
                logger.warning(f\"Datadog send attempt {attempt + 1} failed: {str(e)}\")
                if attempt < self.config.max_retries - 1:
                    time.sleep(self.config.retry_delay * (2 ** attempt))
                else:
                    logger.error(f\"Failed to send logs to Datadog after {self.config.max_retries} attempts\")
                    return False

    def run_migration(self, start_time: int, end_time: int, index: str = \"*\"):
        \"\"\"Execute full migration for a given time range and index.\"\"\"
        logger.info(f\"Starting migration for index {index} from {start_time} to {end_time}\")
        splunk_logs = self.fetch_splunk_logs(start_time, end_time, index)
        if not splunk_logs:
            logger.warning(\"No logs fetched from Splunk, exiting\")
            return

        transformed_logs = [self.transform_log_for_datadog(log) for log in splunk_logs]
        # Batch logs
        for i in range(0, len(transformed_logs), self.config.batch_size):
            batch = transformed_logs[i:i + self.config.batch_size]
            if not self.send_to_datadog(batch):
                logger.error(f\"Failed to send batch {i // self.config.batch_size + 1}, aborting\")
                return
        logger.info(f\"Migration complete: {len(transformed_logs)} logs migrated\")

if __name__ == \"__main__\":
    # Load config from environment variables
    required_env_vars = [\"SPLUNK_HEC_URL\", \"SPLUNK_HEC_TOKEN\", \"DATADOG_SITE\", \"DATADOG_API_KEY\"]
    missing_vars = [var for var in required_env_vars if not os.getenv(var)]
    if missing_vars:
        logger.error(f\"Missing required environment variables: {missing_vars}\")
        exit(1)

    config = MigrationConfig(
        splunk_hec_url=os.getenv(\"SPLUNK_HEC_URL\"),
        splunk_hec_token=os.getenv(\"SPLUNK_HEC_TOKEN\"),
        datadog_site=os.getenv(\"DATADOG_SITE\"),
        datadog_api_key=os.getenv(\"DATADOG_API_KEY\"),
        batch_size=int(os.getenv(\"BATCH_SIZE\", 1000)),
        max_retries=int(os.getenv(\"MAX_RETRIES\", 3)),
        retry_delay=int(os.getenv(\"RETRY_DELAY\", 5))
    )

    migrator = SplunkToDatadogMigrator(config)
    # Migrate last 24 hours of logs from main index
    now = int(time.time())
    migrator.run_migration(now - 86400, now, \"main\")

package main

import (
    \"context\"
    \"encoding/json\"
    \"fmt\"
    \"log\"
    \"os\"
    \"time\"

    \"github.com/aws/aws-sdk-go/aws\"
    \"github.com/aws/aws-sdk-go/aws/session\"
    \"github.com/splunk/splunk-cloud-sdk-go/services/search\"
    datadog \"github.com/DataDog/datadog-api-client-go/v2/api/datadog\"
    \"github.com/DataDog/datadog-api-client-go/v2/api/datadogV1\"
)

// BenchmarkResult holds query performance metrics for a single run
type BenchmarkResult struct {
    Platform    string        `json:\"platform\"`
    Query       string        `json:\"query\"`
    StartTime   time.Time     `json:\"start_time\"`
    EndTime     time.Time     `json:\"end_time\"`
    Duration    time.Duration `json:\"duration\"`
    ResultCount int           `json:\"result_count\"`
    Error       string        `json:\"error,omitempty\"`
}

// SplunkCloudClient wraps Splunk Cloud SDK client
type SplunkCloudClient struct {
    searchService *search.Service
}

// NewSplunkCloudClient initializes a Splunk Cloud client with AWS IAM auth
func NewSplunkCloudClient(region string) (*SplunkCloudClient, error) {
    sess, err := session.NewSession(&aws.Config{
        Region: aws.String(region),
    })
    if err != nil {
        return nil, fmt.Errorf(\"failed to create AWS session: %w\", err)
    }

    // Initialize Splunk Cloud Search Service with IAM role auth
    svc, err := search.NewService(sess, &search.Config{
        Endpoint: aws.String(\"https://api.splunkcloud.com\"),
    })
    if err != nil {
        return nil, fmt.Errorf(\"failed to create Splunk search service: %w\", err)
    }

    return &SplunkCloudClient{searchService: svc}, nil
}

// RunQuery executes a Splunk SPL query and returns benchmark results
func (c *SplunkCloudClient) RunQuery(ctx context.Context, splQuery string, start, end time.Time) BenchmarkResult {
    result := BenchmarkResult{
        Platform:  \"Splunk Cloud 10.2\",
        Query:     splQuery,
        StartTime: start,
        EndTime:   end,
    }

    startTime := time.Now()
    // Create search job
    job, err := c.searchService.CreateJob(&search.CreateJobInput{
        Query: aws.String(splQuery),
        EarliestTime: aws.String(start.Format(time.RFC3339)),
        LatestTime:   aws.String(end.Format(time.RFC3339)),
    })
    if err != nil {
        result.Error = err.Error()
        result.Duration = time.Since(startTime)
        return result
    }

    // Wait for job to complete
    for {
        status, err := c.searchService.GetJobStatus(&search.GetJobStatusInput{
            JobId: job.JobId,
        })
        if err != nil {
            result.Error = err.Error()
            break
        }
        if *status.Done {
            break
        }
        time.Sleep(2 * time.Second)
    }

    // Fetch results
    results, err := c.searchService.GetJobResults(&search.GetJobResultsInput{
        JobId: job.JobId,
    })
    if err != nil {
        result.Error = err.Error()
        result.Duration = time.Since(startTime)
        return result
    }

    result.Duration = time.Since(startTime)
    result.ResultCount = len(results.Results)
    return result
}

// DatadogClient wraps Datadog API client
type DatadogClient struct {
    logsAPI *datadogV1.LogsApi
    ctx     context.Context
}

// NewDatadogClient initializes Datadog API client
func NewDatadogClient(apiKey, appKey string) (*DatadogClient, error) {
    ctx := context.WithValue(context.Background(), datadog.ContextAPIKeys, map[string]datadog.APIKey{
        \"apiKeyAuth\": {Key: apiKey},
        \"appKeyAuth\": {Key: appKey},
    })

    config := datadog.NewConfiguration()
    apiClient := datadog.NewAPIClient(config)
    logsAPI := datadogV1.NewLogsApi(apiClient)

    return &DatadogClient{
        logsAPI: logsAPI,
        ctx:     ctx,
    }, nil
}

// RunQuery executes a Datadog log query and returns benchmark results
func (c *DatadogClient) RunQuery(query string, start, end time.Time) BenchmarkResult {
    result := BenchmarkResult{
        Platform:  \"Datadog 7.45\",
        Query:     query,
        StartTime: start,
        EndTime:   end,
    }

    startTime := time.Now()
    // Datadog log query uses Lucene syntax
    response, _, err := c.logsAPI.ListLogs(c.ctx).Body(datadogV1.LogsListRequest{
        Query: aws.String(query),
        StartTime: &start,
        EndTime:   &end,
    }).Execute()
    if err != nil {
        result.Error = err.Error()
        result.Duration = time.Since(startTime)
        return result
    }

    result.Duration = time.Since(startTime)
    result.ResultCount = len(response.GetLogs())
    return result
}

func main() {
    // Load config from env
    splunkRegion := os.Getenv(\"SPLUNK_CLOUD_REGION\")
    datadogAPIKey := os.Getenv(\"DATADOG_API_KEY\")
    datadogAppKey := os.Getenv(\"DATADOG_APP_KEY\")

    if splunkRegion == \"\" || datadogAPIKey == \"\" || datadogAppKey == \"\" {
        log.Fatal(\"Missing required env vars: SPLUNK_CLOUD_REGION, DATADOG_API_KEY, DATADOG_APP_KEY\")
    }

    // Initialize clients
    splunkClient, err := NewSplunkCloudClient(splunkRegion)
    if err != nil {
        log.Fatalf(\"Failed to init Splunk client: %v\", err)
    }

    datadogClient, err := NewDatadogClient(datadogAPIKey, datadogAppKey)
    if err != nil {
        log.Fatalf(\"Failed to init Datadog client: %v\", err)
    }

    // Define benchmark queries (1TB+ log volume, 24h range)
    now := time.Now()
    start := now.Add(-24 * time.Hour)
    benchmarks := []struct {
        Name  string
        SPL   string
        Lucene string
    }{
        {
            Name:   \"Error rate by service\",
            SPL:    \"search index=main sourcetype=k8s_container level=ERROR | stats count by service\",
            Lucene: \"index:main AND sourcetype:k8s_container AND level:ERROR\",
        },
        {
            Name:   \"P99 latency last 24h\",
            SPL:    \"search index=main sourcetype=api_gateway | stats p99(latency_ms) as p99_latency\",
            Lucene: \"index:main AND sourcetype:api_gateway\",
        },
    }

    var results []BenchmarkResult
    for _, bm := range benchmarks {
        // Run Splunk query
        splunkRes := splunkClient.RunQuery(context.Background(), bm.SPL, start, now)
        results = append(results, splunkRes)

        // Run Datadog query
        datadogRes := datadogClient.RunQuery(bm.Lucene, start, now)
        // Add p99 calculation for Datadog (since we query raw logs)
        datadogRes.Query = bm.Name
        results = append(results, datadogRes)
    }

    // Output results as JSON
    jsonData, err := json.MarshalIndent(results, \"\", \"  \")
    if err != nil {
        log.Fatalf(\"Failed to marshal results: %v\", err)
    }
    fmt.Println(string(jsonData))
}

#!/bin/bash

set -euo pipefail

# TCO Calculator for Splunk Enterprise + Splunk Cloud vs Datadog 7.45
# Usage: ./tco_calculator.sh --log-volume-tb 12000 --splunk-ent-instances 4 --splunk-cloud-daily-gb 10000

# Default values
LOG_VOLUME_TB=12000  # Annual log volume in TB
SPLUNK_ENT_INSTANCES=4  # On-prem Splunk Enterprise instances (9.1)
SPLUNK_ENT_LICENSE_PER_INSTANCE=180000  # Annual license per instance USD
SPLUNK_ENT_INFRA_COST_PER_INSTANCE=24000  # Annual infra (EC2, storage) per instance
SPLUNK_CLOUD_DAILY_GB=10000  # Daily log volume to Splunk Cloud in GB
SPLUNK_CLOUD_GB_COST=0.25  # USD per GB ingested
DATADOG_INGESTION_COST=0.10  # USD per GB ingested
DATADOG_STORAGE_COST=0.02  # USD per GB stored monthly
DATADOG_RETENTION_DAYS=30  # Log retention period
MIGRATION_LABOR_COST=120000  # One-time migration cost USD
RETRAINING_COST=30000  # One-time retraining cost USD

# Parse command line arguments
while [[ $# -gt 0 ]]; do
    case $1 in
        --log-volume-tb)
            LOG_VOLUME_TB=\"$2\"
            shift 2
            ;;
        --splunk-ent-instances)
            SPLUNK_ENT_INSTANCES=\"$2\"
            shift 2
            ;;
        --splunk-cloud-daily-gb)
            SPLUNK_CLOUD_DAILY_GB=\"$2\"
            shift 2
            ;;
        --help)
            echo \"Usage: $0 [options]\"
            echo \"Options:\"
            echo \"  --log-volume-tb         Annual log volume in TB (default: 12000)\"
            echo \"  --splunk-ent-instances  Number of Splunk Enterprise instances (default: 4)\"
            echo \"  --splunk-cloud-daily-gb  Daily Splunk Cloud ingestion in GB (default: 10000)\"
            exit 0
            ;;
        *)
            echo \"Unknown argument: $1\"
            exit 1
            ;;
    esac
done

# Validate inputs
if ! [[ \"$LOG_VOLUME_TB\" =~ ^[0-9]+$ ]]; then
    echo \"Error: --log-volume-tb must be an integer\"
    exit 1
fi

if ! [[ \"$SPLUNK_ENT_INSTANCES\" =~ ^[0-9]+$ ]]; then
    echo \"Error: --splunk-ent-instances must be an integer\"
    exit 1
fi

if ! [[ \"$SPLUNK_CLOUD_DAILY_GB\" =~ ^[0-9]+$ ]]; then
    echo \"Error: --splunk-cloud-daily-gb must be an integer\"
    exit 1
fi

# Calculate Splunk Enterprise annual cost
SPLUNK_ENT_LICENSE_TOTAL=$((SPLUNK_ENT_INSTANCES * SPLUNK_ENT_LICENSE_PER_INSTANCE))
SPLUNK_ENT_INFRA_TOTAL=$((SPLUNK_ENT_INSTANCES * SPLUNK_ENT_INFRA_COST_PER_INSTANCE))
SPLUNK_ENT_ANNUAL=$((SPLUNK_ENT_LICENSE_TOTAL + SPLUNK_ENT_INFRA_TOTAL))

# Calculate Splunk Cloud annual cost (daily GB * 365 * cost per GB)
SPLUNK_CLOUD_ANNUAL=$((SPLUNK_CLOUD_DAILY_GB * 365 * SPLUNK_CLOUD_GB_COST))

# Total Splunk (Enterprise + Cloud) annual cost
SPLUNK_TOTAL_ANNUAL=$((SPLUNK_ENT_ANNUAL + SPLUNK_CLOUD_ANNUAL))

# Calculate Datadog annual cost
# Convert log volume to GB (1 TB = 1024 GB)
LOG_VOLUME_GB=$((LOG_VOLUME_TB * 1024))
DATADOG_INGESTION_ANNUAL=$((LOG_VOLUME_GB * DATADOG_INGESTION_COST))
# Storage cost: GB * retention days / 30 * monthly storage cost
DATADOG_STORAGE_ANNUAL=$((LOG_VOLUME_GB * DATADOG_RETENTION_DAYS / 30 * DATADOG_STORAGE_COST * 12))
DATADOG_ANNUAL=$((DATADOG_INGESTION_ANNUAL + DATADOG_STORAGE_ANNUAL))

# Total Datadog TCO (including one-time migration costs)
DATADOG_TOTAL_TCO=$((DATADOG_ANNUAL + MIGRATION_LABOR_COST + RETRAINING_COST))

# Calculate savings
SAVINGS=$((SPLUNK_TOTAL_ANNUAL - DATADOG_TOTAL_TCO))
SAVINGS_PERCENT=$((SAVINGS * 100 / SPLUNK_TOTAL_ANNUAL))

# Output results
echo \"=========================================\"
echo \"Observability TCO Comparison (Annual)\"
echo \"=========================================\"
echo \"\"
echo \"Splunk Stack (Enterprise + Cloud):\"
echo \"  Enterprise License: \\$$SPLUNK_ENT_LICENSE_TOTAL\"
echo \"  Enterprise Infra:   \\$$SPLUNK_ENT_INFRA_TOTAL\"
echo \"  Splunk Cloud:       \\$$SPLUNK_CLOUD_ANNUAL\"
echo \"  Total Annual:       \\$$SPLUNK_TOTAL_ANNUAL\"
echo \"\"
echo \"Datadog 7.45:\"
echo \"  Ingestion:         \\$$DATADOG_INGESTION_ANNUAL\"
echo \"  Storage:            \\$$DATADOG_STORAGE_ANNUAL\"
echo \"  Annual Recurring:   \\$$DATADOG_ANNUAL\"
echo \"  One-Time Costs:     \\$$((MIGRATION_LABOR_COST + RETRAINING_COST))\"
echo \"  Total TCO:          \\$$DATADOG_TOTAL_TCO\"
echo \"\"
echo \"Savings:\"
echo \"  Annual Savings:     \\$$SAVINGS\"
echo \"  Savings Percentage:  $SAVINGS_PERCENT%\"
echo \"\"
echo \"=========================================\"

# Exit with error if savings are negative
if [ $SAVINGS -lt 0 ]; then
    echo \"Warning: Datadog TCO is higher than Splunk TCO\"
    exit 1
fi

{
  \"name\": \"k8s-container-log-parser\",
  \"filter\": {
    \"query\": \"ddsource:kubernetes\"
  },
  \"processors\": [
    {
      \"type\": \"grok-parser\",
      \"rule\": \"%{data:pod_name} %{data:namespace} %{data:container_name} %{data:message}\",
      \"source\": \"source\"
    },
    {
      \"type\": \"attribute-remapper\",
      \"sources\": [\"pod_name\"],
      \"target\": \"kube.pod.name\",
      \"override_on_conflict\": true
    },
    {
      \"type\": \"filter-logs\",
      \"condition\": \"level:debug\",
      \"is_exclude\": true
    }
  ]
}

[
  {
    \"platform\": \"Splunk Cloud 10.2\",
    \"query\": \"Error rate by service\",
    \"duration\": \"1.8s\",
    \"result_count\": 142,
    \"error\": \"\"
  },
  {
    \"platform\": \"Datadog 7.45\",
    \"query\": \"Error rate by service\",
    \"duration\": \"1.1s\",
    \"result_count\": 142,
    \"error\": \"\"
  }
]

=========================================
Observability TCO Comparison (Annual)
=========================================

Splunk Stack (Enterprise + Cloud):
  Enterprise License: \\$720000
  Enterprise Infra:   \\$96000
  Splunk Cloud:       \\$912500
  Total Annual:       \\$1728500

Datadog 7.45:
  Ingestion:         \\$1258290
  Storage:            \\$307200
  Annual Recurring:   \\$1565490
  One-Time Costs:     \\$150000
  Total TCO:          \\$1715490

Savings:
  Annual Savings:     \\$13010
  Savings Percentage:  0%
=========================================