Central log management

service:api AND (level:>=error OR "payment failed") -host:staging*
attr.customer_id:* AND trace_id:0123456789abcdef0123456789abcdef
message:~"timeout|connection refused"

POST /api/v1/logs/ingest HTTP/1.1
Host: metribeacon.com
Authorization: Bearer <SOURCE_KEY>
Content-Type: application/json

{"batch_id":"batch-1","records":[{"id":"entry-1","level":"info","message":"worker ready","attributes":{"zero":0,"ready":true}}]}

[SERVICE]
    Flush 1
    Log_Level error
    Parsers_File parsers.conf

[INPUT]
    Name tail
    Path /var/log/my-application/*.log
    Parser application_json
    DB /var/lib/fluent-bit/metribeacon-positions.db
    Read_from_Head false
    Rotate_Wait 30
    Buffer_Max_Size 64k
    Skip_Long_Lines Off
    Mem_Buf_Limit 16M
    Tag application

[OUTPUT]
    Name http
    Match application
    Host metribeacon.com
    Port 443
    URI /api/v1/logs/ingest
    Format json_lines
    Json_Date_Key time
    Json_Date_Format iso8601
    Header Authorization Bearer ${METRIBEACON_LOG_KEY}
    tls On
    tls.verify On
    compress gzip
    Workers 1
    Retry_Limit no_limits

[PARSER]
    Name application_json
    Format json
    Time_Key time
    Time_Format %Y-%m-%dT%H:%M:%S.%L%z
    Time_Keep On

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 127.0.0.1:4317
      http:
        endpoint: 127.0.0.1:4318
processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 96
    spike_limit_mib: 16
  batch:
    timeout: 1s
    send_batch_size: 100
    send_batch_max_size: 100
exporters:
  otlphttp/metribeacon:
    endpoint: https://metribeacon.com/api/v1/otlp
    headers:
      Authorization: "Bearer ${env:METRIBEACON_LOG_KEY}"
    compression: gzip
    timeout: 5s
    retry_on_failure:
      enabled: true
      initial_interval: 1s
      max_interval: 30s
      max_elapsed_time: 300s
    sending_queue:
      enabled: true
      num_consumers: 1
      queue_size: 100
service:
  telemetry:
    logs:
      level: error
  pipelines:
    logs:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [otlphttp/metribeacon]

{
  "id": "00000000-0000-4000-8000-000000000001",
  "kind": "file",
  "path": "/var/log/my-app/application.log",
  "multiline": "^\\d{4}-\\d{2}-\\d{2}",
  "policy": {
    "masks": []
  }
}

sudo metribeacon agent logs add --source-file ./log-source.json --key-file ./private-log.key
sudo metribeacon agent logs remove --source-file ./log-source.json

#!/usr/bin/env python3
"""Forward one bounded NDJSON stream; retries retain the same batch and record IDs."""
import argparse
import json
import os
from pathlib import Path
import sys
import time
from urllib.error import HTTPError, URLError
from urllib.request import Request, HTTPRedirectHandler, build_opener
import uuid

class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, request, fp, code, message, headers, target):
        return None

def batches(stream):
    for line in stream:
        if len(line)>65536:raise ValueError('log_record_limit')
        record=json.loads(line)
        if not isinstance(record,dict):raise ValueError('invalid_log_record')
        record.setdefault('id',str(uuid.uuid4()))
        yield {'batch_id':str(uuid.uuid4()),'records':[record]}

def send(batch, token, provider, opener=None):
    raw=json.dumps(batch,separators=(',',':')).encode()
    if len(raw)>1048576:raise ValueError('log_body_limit')
    endpoint='https://metribeacon.com/api/v1/logs/integrations/'+provider
    opener=opener or build_opener(NoRedirect())
    for attempt in range(8):
        request=Request(endpoint,data=raw,headers={'Authorization':'Bearer '+token,'Content-Type':'application/json'},method='POST')
        delay=min(30,2**attempt)
        try:
            with opener.open(request,timeout=10) as response:
                if response.status in (200,202):return
                raise ValueError('log_ingest_rejected')
        except HTTPError as error:
            if error.code not in (429,502,503,504):raise ValueError('log_ingest_rejected') from None
            retry=error.headers.get('Retry-After','')
            if retry.isdecimal():delay=min(300,int(retry))
            error.close()
        except (URLError,TimeoutError):pass
        time.sleep(delay)
    raise ValueError('log_ingest_unconfirmed')

def main():
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--provider',choices=['http','railway','fly','render','supabase','netlify','cloudflare','cloudwatch'],default='http')
    parser.add_argument('--key-file',required=True)
    args=parser.parse_args()
    path=Path(args.key_file)
    if path.is_symlink() or path.stat().st_mode & 0o077:raise ValueError('private_key_file_required')
    token=path.read_text().strip()
    if not token.startswith('mbg_'):raise ValueError('invalid_log_key')
    for batch in batches(iter(lambda:sys.stdin.buffer.readline(65538),b'')):
        send(batch,token,args.provider)

if __name__=='__main__':
    try:main()
    except (ValueError,OSError):
        print('log_forwarding_stopped',file=sys.stderr)
        raise SystemExit(1)

[sinks.metri]
type = "http"
inputs = ["log_json"]
uri = "https://metribeacon.com/api/v1/logs/integrations/fly"
method = "post"
encoding.codec = "json"
framing.method = "newline_delimited"
compression = "gzip"
auth.strategy = "bearer"
auth.token = "${METRIBEACON_LOG_KEY}"
batch.max_bytes = 900000
batch.max_events = 1000
batch.timeout_secs = 1
buffer.type = "memory"
buffer.max_events = 1000
buffer.when_full = "block"
request.concurrency = 1
request.timeout_secs = 10

const MAX_INPUT=5_000_000;
async function bounded(stream,max){const reader=stream.getReader();const chunks=[];let length=0;for(;;){const {done,value}=await reader.read();if(done)break;length+=value.length;if(length>max){await reader.cancel();throw new Error("log_body_limit")};chunks.push(value)};const all=new Uint8Array(length);let offset=0;for(const chunk of chunks){all.set(chunk,offset);offset+=chunk.length};return all}
export async function forward(request,env,fetcher=fetch){
 if(request.method!=="POST")return new Response(null,{status:405});
 if(!env.METRIBEACON_LOG_KEY||request.headers.get("Authorization")!=="Bearer "+env.METRIBEACON_LOG_KEY)return new Response(null,{status:401});
 try{
  let bytes=await bounded(request.body,MAX_INPUT);
  if(request.headers.get("Content-Encoding")==="gzip")bytes=await bounded(new Blob([bytes]).stream().pipeThrough(new DecompressionStream("gzip")),MAX_INPUT);
  const text=new TextDecoder("utf-8",{fatal:true}).decode(bytes);
  if(text.trim()==='{"content":"tests"}')return new Response(null,{status:200});
  const records=text.trim().split("\n").filter(Boolean).map(line=>JSON.parse(line));
  if(records.length>1000)return new Response(null,{status:413});
  const batches=[];let batch=[],size=2;
  for(const record of records){const encoded=JSON.stringify(record),length=new TextEncoder().encode(encoded).length;if(length>65536)return new Response(null,{status:413});if(batch.length&&size+length+1>900000){batches.push(batch);batch=[];size=2};batch.push(record);size+=length+1};if(batch.length)batches.push(batch);
  for(const batch of batches){const response=await fetcher("https://metribeacon.com/api/v1/logs/integrations/cloudflare",{method:"POST",redirect:"error",headers:{Authorization:"Bearer "+env.METRIBEACON_LOG_KEY,"Content-Type":"application/json"},body:JSON.stringify(batch)});await response.body?.cancel();if(!response.ok)return new Response(null,{status:response.status===429?429:response.status>=500?503:response.status,headers:{"Retry-After":response.headers.get("Retry-After")??"5"}})}
  return new Response(null,{status:200});
 }catch(error){return new Response(null,{status:error instanceof Error&&error.message==="log_body_limit"?413:422})}
}
export default {fetch:forward};

{
  "name": "metribeacon-log-forwarder",
  "main": "cloudflare-worker.mjs",
  "compatibility_date": "2026-10-18",
  "observability": {
    "enabled": false
  }
}

{
  "AWSTemplateFormatVersion": "2010-09-09",
  "Parameters": {
    "SourceLogGroup": {
      "Type": "String"
    },
    "LogKeySecretArn": {
      "Type": "String",
      "AllowedPattern": "arn:aws[a-z-]*:secretsmanager:[a-z0-9-]+:[0-9]{12}:secret:.+"
    }
  },
  "Resources": {
    "ForwarderRole": {
      "Type": "AWS::IAM::Role",
      "Properties": {
        "AssumeRolePolicyDocument": {
          "Version": "2012-10-17",
          "Statement": [
            {
              "Effect": "Allow",
              "Principal": {
                "Service": "lambda.amazonaws.com"
              },
              "Action": "sts:AssumeRole"
            }
          ]
        },
        "Policies": [
          {
            "PolicyName": "ReadLogWriteKey",
            "PolicyDocument": {
              "Version": "2012-10-17",
              "Statement": [
                {
                  "Effect": "Allow",
                  "Action": "secretsmanager:GetSecretValue",
                  "Resource": {
                    "Ref": "LogKeySecretArn"
                  }
                }
              ]
            }
          }
        ]
      }
    },
    "Forwarder": {
      "Type": "AWS::Lambda::Function",
      "Properties": {
        "Runtime": "python3.13",
        "Handler": "index.handler",
        "MemorySize": 128,
        "Timeout": 60,
        "ReservedConcurrentExecutions": 2,
        "Role": {
          "Fn::GetAtt": [
            "ForwarderRole",
            "Arn"
          ]
        },
        "Environment": {
          "Variables": {
            "METRIBEACON_LOG_SECRET_ARN": {
              "Ref": "LogKeySecretArn"
            }
          }
        },
        "Code": {
          "ZipFile": "import base64\nimport gzip\nimport io\nimport json\nimport os\nfrom urllib.request import Request, HTTPRedirectHandler, build_opener\nimport boto3\n\nclass NoRedirect(HTTPRedirectHandler):\n    def redirect_request(self, request, fp, code, message, headers, target):\n        return None\n\ndef split_event(event):\n    encoded=base64.b64decode(event['awslogs']['data'],validate=True)\n    with gzip.GzipFile(fileobj=io.BytesIO(encoded)) as compressed:\n        raw=compressed.read(16*1024*1024+1)\n    if len(raw)>16*1024*1024:raise ValueError('cloudwatch_batch_limit')\n    data=json.loads(raw)\n    if data.get('messageType')=='CONTROL_MESSAGE':return []\n    if data.get('logGroup')=='/aws/lambda/'+os.environ.get('AWS_LAMBDA_FUNCTION_NAME',''):\n        raise ValueError('recursive_log_subscription')\n    batches=[];records=[];size=0\n    for item in data.get('logEvents',[]):\n        record={'id':item['id'],'timestamp':item['timestamp'],'message':item['message'],'service':data.get('logGroup',''),'host':data.get('logStream','')}\n        length=len(json.dumps(record,separators=(',',':')).encode())\n        if length>65536:raise ValueError('cloudwatch_record_limit')\n        if records and (len(records)>=1000 or size+length>900000):batches.append(records);records=[];size=0\n        records.append(record);size+=length\n    if records:batches.append(records)\n    return batches\n\ndef handler(event, context):\n    batches=split_event(event)\n    if not batches:return {'accepted':0}\n    token=boto3.client('secretsmanager').get_secret_value(SecretId=os.environ['METRIBEACON_LOG_SECRET_ARN'])['SecretString'].strip()\n    if not token.startswith('mbg_'):raise ValueError('invalid_log_key')\n    accepted=0\n    for records in batches:\n        raw=json.dumps(records,separators=(',',':')).encode()\n        request=Request('https://metribeacon.com/api/v1/logs/integrations/cloudwatch',data=raw,headers={'Authorization':'Bearer '+token,'Content-Type':'application/json'},method='POST')\n        try:\n            with build_opener(NoRedirect()).open(request,timeout=10) as response:\n                if response.status not in (200,202):raise ValueError('log_ingest_unconfirmed')\n        except Exception:\n            raise ValueError('log_ingest_unconfirmed') from None\n        accepted+=len(records)\n    return {'accepted':accepted}\n"
        }
      }
    },
    "InvokeFromLogs": {
      "Type": "AWS::Lambda::Permission",
      "Properties": {
        "Action": "lambda:InvokeFunction",
        "FunctionName": {
          "Ref": "Forwarder"
        },
        "Principal": {
          "Fn::Sub": "logs.${AWS::Region}.amazonaws.com"
        },
        "SourceAccount": {
          "Ref": "AWS::AccountId"
        },
        "SourceArn": {
          "Fn::Sub": "arn:${AWS::Partition}:logs:${AWS::Region}:${AWS::AccountId}:log-group:${SourceLogGroup}:*"
        }
      }
    },
    "Subscription": {
      "Type": "AWS::Logs::SubscriptionFilter",
      "DependsOn": "InvokeFromLogs",
      "Properties": {
        "DestinationArn": {
          "Fn::GetAtt": [
            "Forwarder",
            "Arn"
          ]
        },
        "FilterPattern": "",
        "LogGroupName": {
          "Ref": "SourceLogGroup"
        }
      }
    }
  },
  "Outputs": {
    "ForwarderArn": {
      "Value": {
        "Fn::GetAtt": [
          "Forwarder",
          "Arn"
        ]
      }
    }
  }
}