feat: add metric counting using OTL

Dockerfile.launchContract

@@ -3,6 +3,6 @@ FROM ghcr.io/foundry-rs/foundry as contractLauncher
 # Build and test the source code
 COPY ./contracts .
 RUN forge install foundry-rs/forge-std --no-git
-RUN forge install openzeppelin/openzeppelin-contracts --no-git
-RUN forge build --use 0.8.17
-ENTRYPOINT forge create --contracts ./src/CeramicAnchorServiceV2.sol --private-key $ETH_WALLET_PK --rpc-url $ETH_RPC_URL CeramicAnchorServiceV2
+RUN forge install openzeppelin/openzeppelin-contracts@v4.8.0 --no-git
+RUN forge build --use 0.8.13
+ENTRYPOINT forge create --contracts ./src/CeramicAnchorServiceV2.sol --private-key $ETH_WALLET_PK --rpc-url $ETH_RPC_URL CeramicAnchorServiceV2

cmd/cas/main.go

@@ -15,6 +15,7 @@ import (
 	"github.com/ceramicnetwork/go-cas/common/aws/ddb"
 	"github.com/ceramicnetwork/go-cas/common/aws/queue"
 	"github.com/ceramicnetwork/go-cas/common/db"
+	"github.com/ceramicnetwork/go-cas/common/metrics"
 	"github.com/ceramicnetwork/go-cas/common/notifs"
 	"github.com/ceramicnetwork/go-cas/models"
 	"github.com/ceramicnetwork/go-cas/services"
@@ -67,6 +68,13 @@ func main() {
 		log.Fatalf("failed to create discord handler: %v", err)
 	}
 
+	collectorHost := os.Getenv("COLLECTOR_HOST")
+	metricService, err := metrics.NewMetricService(context.Background(), collectorHost)
+	if err != nil {
+		log.Fatalf("failed to create metric service: %v", err)
+	}
+	defer metricService.Shutdown(context.Background())
+
 	// Flow:
 	// ====
 	// 1. Validation service:
@@ -131,11 +139,11 @@ func main() {
 	queue.NewConsumer(statusQueue, statusService.Status).Start()
 
 	// The Batching Service reads from the Ready queue and posts to the Batch queue
-	batchingService := services.NewBatchingService(batchQueue)
+	batchingService := services.NewBatchingService(batchQueue, metricService)
 	queue.NewConsumer(readyQueue, batchingService.Batch).Start()
 
 	// The Validation Service reads from the Validate queue and posts to the Ready and Status queues
-	validationService := services.NewValidationService(stateDb, readyQueue, statusQueue)
+	validationService := services.NewValidationService(stateDb, readyQueue, statusQueue, metricService)
 	queue.NewConsumer(validateQueue, validationService.Validate).Start()
 
 	// Start the polling services last

common/metrics/otl.go

@@ -0,0 +1,77 @@
+package metrics
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp"
+	"go.opentelemetry.io/otel/exporters/stdout/stdoutmetric"
+	"go.opentelemetry.io/otel/metric"
+	sdk "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/resource"
+
+	"github.com/ceramicnetwork/go-cas/models"
+)
+
+type OtlMetricService struct {
+	meterProvider *sdk.MeterProvider
+	meter         metric.Meter
+}
+
+func NewMetricService(ctx context.Context, collectorHost string) (models.MetricService, error) {
+	var resources *resource.Resource
+	// Resources are a special type of attribute that apply to all spans generated by a process.
+	// These should be used to represent underlying metadata about a process that’s non-ephemeral,
+	// for example, the hostname of a process, or its instance ID.
+	resources, err := resource.New(ctx,
+		resource.WithFromEnv(), // pull attributes from OTEL_RESOURCE_ATTRIBUTES and OTEL_SERVICE_NAME environment variables
+		resource.WithAttributes(attribute.String("environment", os.Getenv("ENV"))),
+	)
+	if err != nil {
+		return &OtlMetricService{}, err
+	}
+
+	var exporter sdk.Exporter
+	if collectorHost != "" {
+		options := otlpmetrichttp.WithEndpoint("%s:4318")
+		exporter, err = otlpmetrichttp.New(ctx, options)
+		if err != nil {
+			return &OtlMetricService{}, err
+		}
+	} else {
+		enc := json.NewEncoder(os.Stdout)
+		enc.SetIndent("", "  ")
+		exporter, err = stdoutmetric.New(
+			stdoutmetric.WithEncoder(enc),
+		)
+		if err != nil {
+			return &OtlMetricService{}, err
+		}
+	}
+
+	meterProvider := sdk.NewMeterProvider(
+		sdk.WithResource(resources),
+		sdk.WithReader(sdk.NewPeriodicReader(exporter)),
+	)
+
+	meter := meterProvider.Meter("go-cas")
+
+	return &OtlMetricService{meter: meter, meterProvider: meterProvider}, nil
+}
+
+func (o OtlMetricService) Shutdown(ctx context.Context) {
+	o.meterProvider.Shutdown(ctx)
+}
+
+func (o OtlMetricService) Count(ctx context.Context, name models.MetricName, val int) error {
+	counter, err := o.meter.Int64Counter(string(name) + "_count")
+	if err != nil {
+		return err
+	}
+
+	counter.Add(ctx, int64(val))
+
+	return nil
+}

docker-compose.yml

@@ -20,6 +20,7 @@ services:
       - PG_DB=anchor_db
       - ANCHOR_BATCH_SIZE=4
       - ANCHOR_BATCH_LINGER=5m
+      - COLLECTOR_HOST=collector
     depends_on:
       localstack:
         condition: service_healthy
@@ -66,8 +67,7 @@ services:
       - DB_PORT=5432
       - DB_USERNAME=root
       - DB_PASSWORD=root
-      - SQS_URL=http://localstack:4566/000000000000/cas-anchor-dev-batch
-      - SQS_QUEUE_USE_POLLING=true
+      - SQS_QUEUE_URL=http://localstack:4566/000000000000/cas-anchor-dev-
       - AWS_ACCOUNT_ID=000000000000
       - AWS_REGION=us-east-1
       - AWS_ACCESS_KEY_ID=.
@@ -90,7 +90,9 @@ services:
       replicas: 1
 
   cas_api:
-    image: ceramicnetwork/ceramic-anchor-service:release-candidate
+    build:
+      context: ../ceramic-anchor-service
+      target: base
     container_name: cas_api
     environment:
       - NODE_ENV=dev
@@ -119,6 +121,11 @@ services:
       - DB_USERNAME=root
       - DB_PASSWORD=root
       - VALIDATE_RECORDS=false
+      - AWS_ACCOUNT_ID=000000000000
+      - AWS_REGION=us-east-1
+      - AWS_ACCESS_KEY_ID=.
+      - AWS_SECRET_ACCESS_KEY=.
+      - SQS_QUEUE_URL=http://localstack:4566/000000000000/cas-anchor-dev-
     depends_on:
       cas_ipfs:
         condition: service_healthy
@@ -245,6 +252,21 @@ services:
     networks:
       - internal
 
+  otel-collector:
+    container_name: collector
+    image: otel/opentelemetry-collector-contrib
+    command: [--config=/etc/otel-collector-config.yaml]
+    volumes:
+      - ./otel-collector-config.yaml:/etc/otel-collector-config.yaml
+    ports:
+      - 1888:1888 # pprof extension
+      - 8888:8888 # Prometheus metrics exposed by the collector
+      - 8889:8889 # Prometheus exporter metrics
+      - 13133:13133 # health_check extension
+      - 4317:4317 # OTLP gRPC receiver
+      - 4318:4318 # OTLP http receiver
+      - 55679:55679 # zpages extension
+
 networks:
   internal:
     driver: bridge

go.mod

@@ -20,6 +20,12 @@ require (
 	github.com/google/uuid v1.3.0
 	github.com/jackc/pgx/v5 v5.1.1
 	github.com/joho/godotenv v1.4.0
+	go.opentelemetry.io/otel v1.16.0
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v0.39.0
+	go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v0.39.0
+	go.opentelemetry.io/otel/metric v1.16.0
+	go.opentelemetry.io/otel/sdk v1.16.0
+	go.opentelemetry.io/otel/sdk/metric v0.39.0
 )
 
 require (
@@ -40,19 +46,32 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.13.9 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sts v1.17.6 // indirect
 	github.com/aws/smithy-go v1.13.5 // indirect
+	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
 	github.com/disgoorg/json v1.0.0 // indirect
 	github.com/disgoorg/log v1.2.0 // indirect
+	github.com/go-logr/logr v1.2.4 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.7.0 // indirect
 	github.com/iancoleman/strcase v0.2.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/sasha-s/go-csync v0.0.0-20210812194225-61421b77c44b // indirect
 	github.com/vektah/gqlparser/v2 v2.5.1 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.16.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric v0.39.0 // indirect
+	go.opentelemetry.io/otel/trace v1.16.0 // indirect
+	go.opentelemetry.io/proto/otlp v0.19.0 // indirect
 	golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 // indirect
 	golang.org/x/exp v0.0.0-20220325121720-054d8573a5d8 // indirect
-	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
+	golang.org/x/mod v0.8.0 // indirect
+	golang.org/x/net v0.8.0 // indirect
 	golang.org/x/sync v0.1.0 // indirect
-	golang.org/x/sys v0.1.0 // indirect
-	golang.org/x/text v0.3.8 // indirect
-	golang.org/x/tools v0.1.12 // indirect
+	golang.org/x/sys v0.8.0 // indirect
+	golang.org/x/text v0.8.0 // indirect
+	golang.org/x/tools v0.6.0 // indirect
+	google.golang.org/genproto v0.0.0-20230306155012-7f2fa6fef1f4 // indirect
+	google.golang.org/grpc v1.55.0 // indirect
+	google.golang.org/protobuf v1.30.0 // indirect
 )