Feat: Observability

Makefile

@@ -1,2 +1,12 @@
 test:
 	docker compose -f docker-compose.test.yaml up toxiproxy integration_test --abort-on-container-exit
+
+example-ocpp-201:
+	docker compose -f example/2.0.1/docker-compose.yml up --build
+
+example-ocpp-16:
+	docker compose -f example/1.6/docker-compose.yml up --build
+
+# Run the example with LGTM stack and observability enabled by default:
+example-ocpp16-observability:
+	METRICS_ENABLED=true docker compose -f example/1.6/docker-compose.yml -f example/docker-compose.observability.yaml up --build

README.md

@@ -14,7 +14,7 @@ There are currently no plans of supporting OCPP-S.
 
 ## Installation
 
-Go version 1.13+ is required.
+Go version 1.22+ is required.
 
 ```sh
 go get github.com/lorenzodonini/ocpp-go
@@ -95,6 +95,7 @@ The websocket package supports configuring ping pong for both endpoints.
 
 By default, the client sends a ping every 54 seconds and waits for a pong for 60 seconds, before timing out.
 The values can be configured as follows:
+
 ```go
 cfg := ws.NewClientTimeoutConfig()
 cfg.PingPeriod = 10 * time.Second
@@ -104,6 +105,7 @@ websocketClient.SetTimeoutConfig(cfg)
 
 By default, the server does not send out any pings and waits for a ping from the client for 60 seconds, before timing out.
 To configure the server to send out pings, the `PingPeriod` and `PongWait` must be set to a value greater than 0:
+
 ```go
 cfg := ws.NewServerTimeoutConfig()
 cfg.PingPeriod = 10 * time.Second

docs/observability.md

@@ -0,0 +1,69 @@
+# Observability in ocpp-go
+
+## Metrics
+
+The library currently supports only websocket and ocpp-j server metrics, which are exported via OpenTelemetry.
+To enable metrics, you need to set the metrics exporter on the server:
+
+```go
+// sets up OTLP metrics exporter
+func setupMetrics(address string) error {
+grpcOpts := []grpc.DialOption{
+grpc.WithTransportCredentials(insecure.NewCredentials()),
+}
+
+client, err := grpc.NewClient(address, grpcOpts...)
+
+if err != nil {
+return errors.Wrap(err, "failed to create gRPC connection to collector")
+}
+
+ctx := context.Background()
+
+exporter, err := otlpmetricgrpc.New(ctx, otlpmetricgrpc.WithGRPCConn(client))
+if err != nil {
+return errors.Wrap(err, "failed to create otlp metric exporter")
+}
+
+resource, err := resource.New(ctx,
+resource.WithAttributes(
+semconv.ServiceNameKey.String("centralSystem-demo"),
+semconv.ServiceVersionKey.String("example"),
+),
+resource.WithFromEnv(),
+resource.WithContainer(),
+resource.WithOS(),
+resource.WithOSType(),
+resource.WithHost(),
+)
+if err != nil {
+return errors.Wrap(err, "failed to create resource")
+}
+
+meterProvider := metricsdk.NewMeterProvider(
+metricsdk.WithReader(
+metricsdk.NewPeriodicReader(
+exporter,
+metricsdk.WithInterval(10*time.Second),
+),
+),
+metricsdk.WithResource(resource),
+)
+
+otel.SetMeterProvider(meterProvider)
+return nil
+}
+```
+
+You can check out the [reference implementation](../example/1.6/cs/central_system_sim.go), and deploy it with:
+
+```bash
+make example-ocpp16-observability
+```
+
+> Note: Deploying the example requires docker and docker compose to be installed.
+
+The deployment will start a central system with metrics enabled and a
+full [observability stack](https://github.com/grafana/docker-otel-lgtm).
+
+You can log in to Grafana at http://localhost:3000 with the credentials `admin/admin`.

example/1.6/cp/charge_point_sim.go

@@ -8,15 +8,14 @@ import (
 	"strconv"
 	"time"
 
-	"github.com/lorenzodonini/ocpp-go/ocpp1.6/logging"
-	"github.com/sirupsen/logrus"
-
 	ocpp16 "github.com/lorenzodonini/ocpp-go/ocpp1.6"
 	"github.com/lorenzodonini/ocpp-go/ocpp1.6/core"
 	"github.com/lorenzodonini/ocpp-go/ocpp1.6/localauth"
+	"github.com/lorenzodonini/ocpp-go/ocpp1.6/logging"
 	"github.com/lorenzodonini/ocpp-go/ocpp1.6/types"
 	"github.com/lorenzodonini/ocpp-go/ocppj"
 	"github.com/lorenzodonini/ocpp-go/ws"
+	"github.com/sirupsen/logrus"
 )
 
 const (

example/1.6/cs/central_system_sim.go

@@ -1,13 +1,22 @@
 package main
 
 import (
+	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"os"
 	"strconv"
 	"time"
 
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
+	metricsdk "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/resource"
+	semconv "go.opentelemetry.io/otel/semconv/v1.10.0"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 
 	ocpp16 "github.com/lorenzodonini/ocpp-go/ocpp1.6"
 	"github.com/lorenzodonini/ocpp-go/ocpp1.6/core"
@@ -28,6 +37,8 @@ const (
 	envVarCaCertificate        = "CA_CERTIFICATE_PATH"
 	envVarServerCertificate    = "SERVER_CERTIFICATE_PATH"
 	envVarServerCertificateKey = "SERVER_CERTIFICATE_KEY_PATH"
+	envVarMetricsEnabled       = "METRICS_ENABLED"
+	envVarMetricsAddress       = "METRICS_ADDRESS"
 )
 
 var log *logrus.Logger
@@ -188,6 +199,54 @@ func exampleRoutine(chargePointID string, handler *CentralSystemHandler) {
 	}
 }
 
+// sets up OTLP metrics exporter
+func setupMetrics(address string) error {
+	grpcOpts := []grpc.DialOption{
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	}
+
+	client, err := grpc.NewClient(address, grpcOpts...)
+
+	if err != nil {
+		return errors.Wrap(err, "failed to create gRPC connection to collector")
+	}
+
+	ctx := context.Background()
+
+	exporter, err := otlpmetricgrpc.New(ctx, otlpmetricgrpc.WithGRPCConn(client))
+	if err != nil {
+		return errors.Wrap(err, "failed to create otlp metric exporter")
+	}
+
+	resource, err := resource.New(ctx,
+		resource.WithAttributes(
+			semconv.ServiceNameKey.String("centralSystem-demo"),
+			semconv.ServiceVersionKey.String("example"),
+		),
+		resource.WithFromEnv(),
+		resource.WithContainer(),
+		resource.WithOS(),
+		resource.WithOSType(),
+		resource.WithHost(),
+	)
+	if err != nil {
+		return errors.Wrap(err, "failed to create resource")
+	}
+
+	meterProvider := metricsdk.NewMeterProvider(
+		metricsdk.WithReader(
+			metricsdk.NewPeriodicReader(
+				exporter,
+				metricsdk.WithInterval(10*time.Second),
+			),
+		),
+		metricsdk.WithResource(resource),
+	)
+
+	otel.SetMeterProvider(meterProvider)
+	return nil
+}
+
 // Start function
 func main() {
 	// Load config from ENV
@@ -198,6 +257,16 @@ func main() {
 	} else {
 		log.Printf("no valid %v environment variable found, using default port", envVarServerPort)
 	}
+
+	// Setup metrics if enabled
+	if t, _ := os.LookupEnv(envVarMetricsEnabled); t == "true" {
+		address, _ := os.LookupEnv(envVarMetricsAddress)
+		if err := setupMetrics(address); err != nil {
+			log.Error(err)
+			return
+		}
+	}
+
 	// Check if TLS enabled
 	t, _ := os.LookupEnv(envVarTls)
 	tlsEnabled, _ := strconv.ParseBool(t)
@@ -207,6 +276,7 @@ func main() {
 	} else {
 		centralSystem = setupCentralSystem()
 	}
+
 	// Support callbacks for all OCPP 1.6 profiles
 	handler := &CentralSystemHandler{chargePoints: map[string]*ChargePointState{}}
 	centralSystem.SetCoreHandler(handler)

example/1.6/docker-compose.tls.yml

@@ -1,4 +1,4 @@
-version: '3'
+version: '3.9'
 services:
   central-system:
     build:
@@ -15,6 +15,8 @@ services:
       - CA_CERTIFICATE_PATH=/usr/local/share/certs/ca.crt
       - SERVER_CERTIFICATE_PATH=/usr/local/share/certs/central-system.crt
       - SERVER_CERTIFICATE_KEY_PATH=/usr/local/share/certs/central-system.key
+      - METRICS_ENABLED=${METRICS_ENABLED:-false}
+      - METRICS_ADDRESS=${METRICS_ADDRESS:-lgtm-stack:4317}
     ports:
       - "443:443"
     networks:

example/1.6/docker-compose.yml

@@ -1,12 +1,14 @@
-version: '3'
+version: '3.9'
 services:
   central-system:
     build:
       context: ../..
-      dockerfile: cs/Dockerfile
+      dockerfile: example/1.6/cs/Dockerfile
     image: ldonini/ocpp1.6-central-system:latest
     container_name: central-system
     environment:
+      - METRICS_ENABLED=${METRICS_ENABLED:-false}
+      - METRICS_ADDRESS=${METRICS_ADDRESS:-lgtm-stack:4317}
       - SERVER_LISTEN_PORT=8887
     ports:
       - "8887:8887"
@@ -19,7 +21,7 @@ services:
         condition: service_started
     build:
       context: ../..
-      dockerfile: cp/Dockerfile
+      dockerfile: example/1.6/cp/Dockerfile
     image: ldonini/ocpp1.6-charge-point:latest
     container_name: charge-point
     environment:

example/2.0.1/csms/csms_sim.go

@@ -1,14 +1,23 @@
 package main
 
 import (
+	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
 	"os"
 	"strconv"
 	"time"
 
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
+	metricsdk "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/resource"
+	semconv "go.opentelemetry.io/otel/semconv/v1.10.0"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 
 	"github.com/lorenzodonini/ocpp-go/ocpp2.0.1"
 	"github.com/lorenzodonini/ocpp-go/ocpp2.0.1/availability"
@@ -31,6 +40,8 @@ const (
 	envVarCaCertificate        = "CA_CERTIFICATE_PATH"
 	envVarServerCertificate    = "SERVER_CERTIFICATE_PATH"
 	envVarServerCertificateKey = "SERVER_CERTIFICATE_KEY_PATH"
+	envVarMetricsEnabled       = "METRICS_ENABLED"
+	envVarMetricsAddress       = "METRICS_ADDRESS"
 )
 
 var log *logrus.Logger
@@ -246,6 +257,54 @@ func exampleRoutine(chargingStationID string, handler *CSMSHandler) {
 	// Finish simulation
 }
 
+// sets up OTLP metrics exporter
+func setupMetrics(address string) error {
+	grpcOpts := []grpc.DialOption{
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	}
+
+	client, err := grpc.NewClient(address, grpcOpts...)
+
+	if err != nil {
+		return errors.Wrap(err, "failed to create gRPC connection to collector")
+	}
+
+	ctx := context.Background()
+
+	exporter, err := otlpmetricgrpc.New(ctx, otlpmetricgrpc.WithGRPCConn(client))
+	if err != nil {
+		return errors.Wrap(err, "failed to create otlp metric exporter")
+	}
+
+	resource, err := resource.New(ctx,
+		resource.WithAttributes(
+			semconv.ServiceNameKey.String("csms-demo"),
+			semconv.ServiceVersionKey.String("example"),
+		),
+		resource.WithFromEnv(),
+		resource.WithContainer(),
+		resource.WithOS(),
+		resource.WithOSType(),
+		resource.WithHost(),
+	)
+	if err != nil {
+		return errors.Wrap(err, "failed to create resource")
+	}
+
+	meterProvider := metricsdk.NewMeterProvider(
+		metricsdk.WithReader(
+			metricsdk.NewPeriodicReader(
+				exporter,
+				metricsdk.WithInterval(10*time.Second),
+			),
+		),
+		metricsdk.WithResource(resource),
+	)
+
+	otel.SetMeterProvider(meterProvider)
+	return nil
+}
+
 // Start function
 func main() {
 	// Load config from ENV
@@ -256,6 +315,16 @@ func main() {
 	} else {
 		log.Printf("no valid %v environment variable found, using default port", envVarServerPort)
 	}
+
+	// Setup metrics if enabled
+	if t, _ := os.LookupEnv(envVarMetricsEnabled); t == "true" {
+		address, _ := os.LookupEnv(envVarMetricsAddress)
+		if err := setupMetrics(address); err != nil {
+			log.Error(err)
+			return
+		}
+	}
+
 	// Check if TLS enabled
 	t, _ := os.LookupEnv(envVarTls)
 	tlsEnabled, _ := strconv.ParseBool(t)
@@ -265,6 +334,7 @@ func main() {
 	} else {
 		csms = setupCentralSystem()
 	}
+
 	// Support callbacks for all OCPP 2.0.1 profiles
 	handler := &CSMSHandler{chargingStations: map[string]*ChargingStationState{}}
 	csms.SetAuthorizationHandler(handler)