feat: implement fluent-bit input plugin for cloudwatch logs (#1)

Author: milkrage

Dockerfile

@@ -0,0 +1,18 @@
+FROM golang:1.24-bookworm as build
+
+WORKDIR /src
+
+COPY go.mod go.sum ./
+RUN go mod download -x
+
+COPY . .
+
+RUN go build -buildmode=c-shared -o cloudwatch-input.so plugin.go
+
+
+FROM ghcr.io/selectel/fluent-bit:2025-04-22
+
+COPY --from=build /src/cloudwatch-input.so /fluent-bit/plugins/
+
+ENTRYPOINT ["/fluent-bit/bin/fluent-bit", "-e", "/fluent-bit/plugins/cloudwatch-input.so"]
+CMD ["-c", "/fluent-bit/etc/fluent-bit.yaml"]

Makefile

@@ -0,0 +1,8 @@
+build:
+	go build -buildmode=c-shared -o cloudwatch-input.so plugin.go
+
+run:
+	fluent-bit -e ./cloudwatch-input.so -c config/fluent-bit.yaml
+
+image:
+	docker build -t ghcr.io/selectel/fluent-bit-cloudwatch-input-plugin:development .

README.md

@@ -0,0 +1,14 @@
+# Fluent Bit input plugin for CloudWatch logs
+
+## Quick start
+
+```bash
+docker run \
+  --name fluent-bit-cloudwatch \
+  --rm \
+  -v ${PWD}/config/fluent-bit.yaml:/fluent-bit/etc/fluent-bit.yaml:ro \
+  -v ${PWD}/sqlite:/var/lib/fluent-bit/cloudwatch/sqlite:rw \
+  -e AWS_ACCESS_KEY_ID=your_access_key \
+  -e AWS_SECRET_ACCESS_KEY=your_secret_key \
+  ghcr.io/selectel/fluent-bit-cloudwatch-input-plugin:latest
+```

config/fluent-bit.yaml

@@ -0,0 +1,24 @@
+service:
+  log_level: debug
+
+pipeline:
+  inputs:
+    - name: cloudwatch-input
+      region: us-east-1
+      endpoint: https://logs.us-east-1.amazonaws.com
+      log_group_name: my-group-name
+      log_stream_name: my-stream-name
+      sqlite_path: /var/lib/fluent-bit/cloudwatch/sqlite/db.sqlite
+
+  filters:
+    - name: lua
+      match: '*'
+      call: split_events
+      code: |
+        function split_events(tag, timestamp, record)
+          return 2, timestamp, record["events"]
+        end
+
+  outputs:
+    - name: stdout
+      match: '*'

go.mod

@@ -0,0 +1,27 @@
+module github.com/selectel/fluent-bit-cloudwatch-input-plugin
+
+go 1.24
+
+require (
+	github.com/aws/aws-sdk-go-v2 v1.36.3
+	github.com/aws/aws-sdk-go-v2/config v1.29.14
+	github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs v1.47.3
+	github.com/mattn/go-sqlite3 v1.14.28
+	github.com/vmihailenco/msgpack/v5 v5.4.1
+)
+
+require (
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.10 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.17.67 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.30 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.34 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.34 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.15 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.25.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.30.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.33.19 // indirect
+	github.com/aws/smithy-go v1.22.3 // indirect
+	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
+)

go.sum

@@ -0,0 +1,44 @@
+github.com/aws/aws-sdk-go-v2 v1.36.3 h1:mJoei2CxPutQVxaATCzDUjcZEjVRdpsiiXi2o38yqWM=
+github.com/aws/aws-sdk-go-v2 v1.36.3/go.mod h1:LLXuLpgzEbD766Z5ECcRmi8AzSwfZItDtmABVkRLGzg=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.10 h1:zAybnyUQXIZ5mok5Jqwlf58/TFE7uvd3IAsa1aF9cXs=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.10/go.mod h1:qqvMj6gHLR/EXWZw4ZbqlPbQUyenf4h82UQUlKc+l14=
+github.com/aws/aws-sdk-go-v2/config v1.29.14 h1:f+eEi/2cKCg9pqKBoAIwRGzVb70MRKqWX4dg1BDcSJM=
+github.com/aws/aws-sdk-go-v2/config v1.29.14/go.mod h1:wVPHWcIFv3WO89w0rE10gzf17ZYy+UVS1Geq8Iei34g=
+github.com/aws/aws-sdk-go-v2/credentials v1.17.67 h1:9KxtdcIA/5xPNQyZRgUSpYOE6j9Bc4+D7nZua0KGYOM=
+github.com/aws/aws-sdk-go-v2/credentials v1.17.67/go.mod h1:p3C44m+cfnbv763s52gCqrjaqyPikj9Sg47kUVaNZQQ=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.30 h1:x793wxmUWVDhshP8WW2mlnXuFrO4cOd3HLBroh1paFw=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.30/go.mod h1:Jpne2tDnYiFascUEs2AWHJL9Yp7A5ZVy3TNyxaAjD6M=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.34 h1:ZK5jHhnrioRkUNOc+hOgQKlUL5JeC3S6JgLxtQ+Rm0Q=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.34/go.mod h1:p4VfIceZokChbA9FzMbRGz5OV+lekcVtHlPKEO0gSZY=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.34 h1:SZwFm17ZUNNg5Np0ioo/gq8Mn6u9w19Mri8DnJ15Jf0=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.34/go.mod h1:dFZsC0BLo346mvKQLWmoJxT+Sjp+qcVR1tRVHQGOH9Q=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 h1:bIqFDwgGXXN1Kpp99pDOdKMTTb5d2KyU5X/BZxjOkRo=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3/go.mod h1:H5O/EsxDWyU+LP/V8i5sm8cxoZgc2fdNR9bxlOFrQTo=
+github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs v1.47.3 h1:3y0jkGtsaZLCg+n73BoSXOAkLFtgmD/+4prXW1pzovc=
+github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs v1.47.3/go.mod h1:uo14VBn5cNk/BPGTPz3kyLBxgpgOObgO8lmz+H7Z4Ck=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.3 h1:eAh2A4b5IzM/lum78bZ590jy36+d/aFLgKF/4Vd1xPE=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.3/go.mod h1:0yKJC/kb8sAnmlYa6Zs3QVYqaC8ug2AbnNChv5Ox3uA=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.15 h1:dM9/92u2F1JbDaGooxTq18wmmFzbJRfXfVfy96/1CXM=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.15/go.mod h1:SwFBy2vjtA0vZbjjaFtfN045boopadnoVPhu4Fv66vY=
+github.com/aws/aws-sdk-go-v2/service/sso v1.25.3 h1:1Gw+9ajCV1jogloEv1RRnvfRFia2cL6c9cuKV2Ps+G8=
+github.com/aws/aws-sdk-go-v2/service/sso v1.25.3/go.mod h1:qs4a9T5EMLl/Cajiw2TcbNt2UNo/Hqlyp+GiuG4CFDI=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.30.1 h1:hXmVKytPfTy5axZ+fYbR5d0cFmC3JvwLm5kM83luako=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.30.1/go.mod h1:MlYRNmYu/fGPoxBQVvBYr9nyr948aY/WLUvwBMBJubs=
+github.com/aws/aws-sdk-go-v2/service/sts v1.33.19 h1:1XuUZ8mYJw9B6lzAkXhqHlJd/XvaX32evhproijJEZY=
+github.com/aws/aws-sdk-go-v2/service/sts v1.33.19/go.mod h1:cQnB8CUnxbMU82JvlqjKR2HBOm3fe9pWorWBza6MBJ4=
+github.com/aws/smithy-go v1.22.3 h1:Z//5NuZCSW6R4PhQ93hShNbyBbn8BWCmCVCt+Q8Io5k=
+github.com/aws/smithy-go v1.22.3/go.mod h1:t1ufH5HMublsJYulve2RKmHDC15xu1f26kHCp/HgceI=
+github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/mattn/go-sqlite3 v1.14.28 h1:ThEiQrnbtumT+QMknw63Befp/ce/nUPgBPMlRFEum7A=
+github.com/mattn/go-sqlite3 v1.14.28/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8=
+github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
+github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
+github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/infra/client/cloudwatch.go

@@ -0,0 +1,61 @@
+package client
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
+
+	"github.com/selectel/fluent-bit-cloudwatch-input-plugin/internal/model"
+)
+
+type Cloudwatch struct {
+	client *cloudwatchlogs.Client
+}
+
+func NewCloudwatchClient(ctx context.Context, region, endpoint string) (*Cloudwatch, error) {
+	cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(region), config.WithBaseEndpoint(endpoint))
+	if err != nil {
+		return nil, fmt.Errorf("failed to load aws config: %w", err)
+	}
+
+	client := &Cloudwatch{
+		client: cloudwatchlogs.NewFromConfig(cfg),
+	}
+
+	return client, nil
+}
+
+func (cw *Cloudwatch) GetLogEvents(
+	ctx context.Context,
+	logGroupName, logStreamName, nextToken string,
+) ([]model.Event, string, error) {
+	params := &cloudwatchlogs.GetLogEventsInput{
+		LogGroupName:  aws.String(logGroupName),
+		LogStreamName: aws.String(logStreamName),
+		StartFromHead: aws.Bool(true),
+	}
+
+	if nextToken != "" {
+		params.NextToken = aws.String(nextToken)
+	}
+
+	resp, err := cw.client.GetLogEvents(ctx, params)
+	if err != nil {
+		return nil, "", fmt.Errorf("failed to complete request: %w", err)
+	}
+
+	events := make([]model.Event, 0, len(resp.Events))
+
+	for _, event := range resp.Events {
+		events = append(events, model.Event{
+			IngestionTime: *event.IngestionTime,
+			Timestamp:     *event.Timestamp,
+			Message:       *event.Message,
+		})
+	}
+
+	return events, *resp.NextForwardToken, nil
+}

internal/infra/storage/sqlite.go

@@ -0,0 +1,58 @@
+package storage
+
+import (
+	"database/sql"
+	"fmt"
+	"net/url"
+
+	_ "github.com/mattn/go-sqlite3"
+)
+
+func NewSQLite(path string) (*sql.DB, error) {
+	params := url.Values{}
+
+	params.Add("cache", "shared")
+	params.Add("mode", "rwc")
+	params.Add("_journal", "WAL")
+
+	conn, err := sql.Open("sqlite3", fmt.Sprintf("file:%s?%s", path, params.Encode()))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create conntection: %w", err)
+	}
+
+	// https://github.com/mattn/go-sqlite3/issues/209
+	conn.SetMaxOpenConns(1)
+
+	err = conn.Ping()
+	if err != nil {
+		return nil, fmt.Errorf("failed to ping connection: %w", err)
+	}
+
+	err = migrateSQLite(conn)
+	if err != nil {
+		return nil, fmt.Errorf("failed to run migrations: %w", err)
+	}
+
+	return conn, nil
+}
+
+func migrateSQLite(db *sql.DB) error {
+	const schema = `
+		CREATE TABLE IF NOT EXISTS state (
+			region          TEXT       NOT NULL,
+			log_group_name  TEXT       NOT NULL,
+			log_stream_name TEXT       NOT NULL,
+			next_token      TEXT       NOT NULL,
+			updated_at      INTEGER    NOT NULL,
+
+			PRIMARY KEY (region, log_group_name, log_stream_name)
+		);
+	`
+
+	_, err := db.Exec(schema)
+	if err != nil {
+		return fmt.Errorf("failed to execute sql query: %w", err)
+	}
+
+	return nil
+}

internal/infra/storage/sqlite/state.go

@@ -0,0 +1,49 @@
+package sqlite
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"time"
+)
+
+type State struct {
+	db *sql.DB
+}
+
+func NewState(db *sql.DB) *State {
+	return &State{
+		db: db,
+	}
+}
+
+func (s *State) GetNextToken(ctx context.Context, region, logGroupName, logStreamName string) (string, error) {
+	const query = `SELECT next_token FROM state WHERE region = ? and log_group_name = ? and log_stream_name = ?`
+
+	var nextToken string
+
+	err := s.db.QueryRowContext(ctx, query, region, logGroupName, logStreamName).Scan(&nextToken)
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return "", fmt.Errorf("failed to execute sql query: %w", err)
+	}
+
+	return nextToken, nil
+}
+
+func (s *State) SetNextToken(ctx context.Context, region, logGroupName, logStreamName, nextToken string) error {
+	const query = `
+		INSERT INTO state (region, log_group_name, log_stream_name, next_token, updated_at)
+		VALUES (?, ?, ?, ?, ?)
+		ON CONFLICT (region, log_group_name, log_stream_name) DO UPDATE SET
+			next_token = excluded.next_token,
+			updated_at = excluded.updated_at
+	`
+
+	_, err := s.db.ExecContext(ctx, query, region, logGroupName, logStreamName, nextToken, time.Now().UTC().Unix())
+	if err != nil {
+		return fmt.Errorf("failed to execute sql query: %w", err)
+	}
+
+	return nil
+}

internal/model/event.go

@@ -0,0 +1,7 @@
+package model
+
+type Event struct {
+	IngestionTime int64  `msgpack:"ingestion_time"`
+	Timestamp     int64  `msgpack:"timestamp"`
+	Message       string `msgpack:"message"`
+}

internal/plugin/adapter/cloudwatch.go

@@ -0,0 +1,11 @@
+package adapter
+
+import (
+	"context"
+
+	"github.com/selectel/fluent-bit-cloudwatch-input-plugin/internal/model"
+)
+
+type Cloudwatch interface {
+	GetLogEvents(ctx context.Context, logGroupName, logStreamName, nextToken string) ([]model.Event, string, error)
+}

internal/plugin/adapter/storage.go

@@ -0,0 +1,8 @@
+package adapter
+
+import "context"
+
+type Storage interface {
+	GetNextToken(ctx context.Context, region, logGroupName, logStreamName string) (string, error)
+	SetNextToken(ctx context.Context, region, logGroupName, logStreamName, nextToken string) error
+}

internal/plugin/plugin.go

@@ -0,0 +1,45 @@
+package plugin
+
+import (
+	"context"
+
+	"github.com/selectel/fluent-bit-cloudwatch-input-plugin/internal/model"
+	"github.com/selectel/fluent-bit-cloudwatch-input-plugin/internal/plugin/adapter"
+)
+
+type Plugin struct {
+	region        string
+	endpoint      string
+	logGroupName  string
+	logStreamName string
+
+	cloudwatch adapter.Cloudwatch
+	storage    adapter.Storage
+}
+
+func NewPlugin(
+	region, endpoint, logGroupName, logStreamName string,
+	cloudwatch adapter.Cloudwatch,
+	storage adapter.Storage,
+) *Plugin {
+	return &Plugin{
+		region:        region,
+		endpoint:      endpoint,
+		logGroupName:  logGroupName,
+		logStreamName: logStreamName,
+		cloudwatch:    cloudwatch,
+		storage:       storage,
+	}
+}
+
+func (p *Plugin) GetLogEvents(ctx context.Context, nextToken string) ([]model.Event, string, error) {
+	return p.cloudwatch.GetLogEvents(ctx, p.logGroupName, p.logStreamName, nextToken)
+}
+
+func (p *Plugin) GetNextToken(ctx context.Context) (string, error) {
+	return p.storage.GetNextToken(ctx, p.region, p.logGroupName, p.logStreamName)
+}
+
+func (p *Plugin) SetNextToken(ctx context.Context, nextToken string) error {
+	return p.storage.SetNextToken(ctx, p.region, p.logGroupName, p.logStreamName, nextToken)
+}