[+] add google cloud pub/sub receiver

cmd/gcp_pubsub_receiver/README.md

@@ -0,0 +1,13 @@
+# GCP Pub/Sub Receiver
+
+A gRPC server that writes metrics received from pgwatch
+to Google cloud pub/sub servers.
+
+- The receiver creates a new topic called `pgwatch` in the provided GCP project.
+- The receiver uses the official pub/sub package for golang which supports Authentication via [Application Default Credentials (ADC)](https://cloud.google.com/docs/authentication/application-default-credentials)
+
+## Usage example
+
+```bash
+go run ./cmd/gcp_pubsub_receiver --port <grpc-server-port-number> --project-id <gcp-project-id>
+```

cmd/gcp_pubsub_receiver/main.go

@@ -0,0 +1,23 @@
+package main
+
+import (
+	"flag"
+	"log"
+
+	"github.com/destrex271/pgwatch3_rpc_server/sinks"
+)
+
+func main() {
+	port := flag.String("port", "", "Port number for the server to listen on.")
+	projectID := flag.String("project-id", "", "GCP Project Id.")
+	flag.Parse()
+
+	server, err := NewPubsubReceiver(*projectID)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	if err := sinks.ListenAndServe(server, *port); err != nil {
+		log.Fatal(err)
+	}
+}

cmd/gcp_pubsub_receiver/pubsub_receiver.go

@@ -0,0 +1,55 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"cloud.google.com/go/pubsub/v2"
+	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
+	"github.com/destrex271/pgwatch3_rpc_server/sinks"
+	"github.com/destrex271/pgwatch3_rpc_server/sinks/pb"
+)
+
+type PubsubReceiver struct {
+	client *pubsub.Client
+	publisher *pubsub.Publisher
+	sinks.SyncMetricHandler
+}
+
+func NewPubsubReceiver(projectID string) (*PubsubReceiver, error) {
+	ctx := context.Background()
+
+	client, err := pubsub.NewClient(ctx, projectID)
+	if err != nil {
+		return nil, err
+	}
+
+	topicName := fmt.Sprintf("projects/%s/topics/pgwatch", projectID)
+	topic, err := client.TopicAdminClient.CreateTopic(ctx, &pubsubpb.Topic{
+		Name: topicName,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	publisher := client.Publisher(topic.GetName())
+	pr := &PubsubReceiver{
+		client: client,
+		publisher: publisher,
+		SyncMetricHandler: sinks.NewSyncMetricHandler(1024),
+	}
+
+	go pr.HandleSyncMetric()
+	return pr, nil
+}
+
+func (r *PubsubReceiver) UpdateMeasurements(ctx context.Context, msg *pb.MeasurementEnvelope) (*pb.Reply, error) {
+	data, err := json.Marshal(msg)
+	if err != nil {
+		return nil, err
+	}
+
+	_ = r.publisher.Publish(ctx, &pubsub.Message{Data: data})
+	return &pb.Reply{Logmsg: "Message published."}, nil
+}

cmd/gcp_pubsub_receiver/pubsub_receiver_test.go

@@ -0,0 +1,114 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"os"
+	"testing"
+
+	"cloud.google.com/go/pubsub/v2"
+	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
+	testutils "github.com/destrex271/pgwatch3_rpc_server/sinks/test_utils"
+	"github.com/stretchr/testify/assert"
+	"github.com/testcontainers/testcontainers-go"
+	tcpubsub "github.com/testcontainers/testcontainers-go/modules/gcloud/pubsub"
+	"github.com/testcontainers/testcontainers-go/wait"
+)
+
+var pubsubContainer *tcpubsub.Container
+
+func TestMain(m *testing.M) {
+	var err error
+	pubsubContainer, err = tcpubsub.Run(
+		context.Background(),
+		"gcr.io/google.com/cloudsdktool/cloud-sdk:367.0.0-emulators",
+		tcpubsub.WithProjectID("pubsub-receiver-test-project"),
+		testcontainers.WithExposedPorts("8085:8085/tcp"),
+		testcontainers.WithWaitStrategy(
+			wait.ForLog("Server started"),
+		),
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	err = os.Setenv("PUBSUB_EMULATOR_HOST", pubsubContainer.URI())
+	if err != nil {
+		panic(err)
+	}
+
+	exitCode := m.Run()	
+
+	if err := testcontainers.TerminateContainer(pubsubContainer); err != nil {
+		log.Printf("failed to terminate container: %s", err)
+	}
+	os.Exit(exitCode)
+}
+
+func TestPubsubReceiver(t *testing.T) {
+	a := assert.New(t)
+
+	psr, err := NewPubsubReceiver(pubsubContainer.ProjectID())
+	a.NoError(err)
+	a.NotNil(psr)
+
+	t.Run("Test Pub/Sub Receiver UpdateMeasurements()", func(t *testing.T) {
+		msg := testutils.GetTestMeasurementEnvelope()
+		reply, err := psr.UpdateMeasurements(context.Background(), msg)
+
+		a.NoError(err)
+		a.Equal(reply.GetLogmsg(), "Message published.")
+
+		// Try read the published message from the Pub/Sub server.
+		sub, err := CreateSubscription(psr)
+		a.NoError(err)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		err = sub.Receive(ctx, func(ctx context.Context, m *pubsub.Message) {
+			var recvd_msg map[string]any
+			err := json.Unmarshal(m.Data, &recvd_msg)
+			a.NoError(err)
+
+			a.Equal(msg.GetDBName(), recvd_msg["DBName"])
+			a.Equal(msg.GetMetricName(), recvd_msg["MetricName"])
+
+			data := recvd_msg["Data"].([]any)
+			for i, item := range msg.GetData() {
+				recvd_item := data[i].(map[string]any)
+				a.Equal(item.AsMap(), recvd_item)
+			}
+
+			m.Ack()
+			// cancel the ctx to force Receive() to return
+			cancel()
+		})
+		a.NoError(err)
+	})
+
+	t.Run("Test calling SyncMetric() from Pub/Sub Receiver", func(t *testing.T) {
+		req := testutils.GetTestRPCSyncRequest()
+		reply, err := psr.SyncMetric(context.Background(), req)
+		a.NoError(err)
+		a.Equal(reply.GetLogmsg(), fmt.Sprintf("gRPC Receiver Synced: DBName %s MetricName %s Operation %s", req.GetDBName(), req.GetMetricName(), "Add"))
+	})
+}
+
+func CreateSubscription(psr *PubsubReceiver) (*pubsub.Subscriber, error){
+	subName := fmt.Sprintf("projects/%s/subscriptions/test-sub", pubsubContainer.ProjectID())
+	topicName := fmt.Sprintf("projects/%s/topics/pgwatch", pubsubContainer.ProjectID())
+
+	subscription, err := psr.client.SubscriptionAdminClient.CreateSubscription(context.Background(),
+		&pubsubpb.Subscription{
+			Name: subName,
+			Topic: topicName,
+		},
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	sub := psr.client.Subscriber(subscription.GetName())
+	return sub, nil
+}

go.mod

@@ -3,6 +3,7 @@ module github.com/destrex271/pgwatch3_rpc_server
 go 1.24.0
 
 require (
+	cloud.google.com/go/pubsub/v2 v2.0.0
 	github.com/ClickHouse/clickhouse-go/v2 v2.28.3
 	github.com/elastic/go-elasticsearch/v8 v8.19.0
 	github.com/marcboeker/go-duckdb v1.8.4
@@ -11,11 +12,17 @@ require (
 	github.com/segmentio/kafka-go v0.4.47
 	github.com/stretchr/testify v1.10.0
 	github.com/testcontainers/testcontainers-go v0.38.0
+	github.com/testcontainers/testcontainers-go/modules/gcloud v0.38.0
 	github.com/testcontainers/testcontainers-go/modules/elasticsearch v0.38.0
 	github.com/testcontainers/testcontainers-go/modules/localstack v0.37.0
 )
 
 require (
+	cloud.google.com/go v0.121.1 // indirect
+	cloud.google.com/go/auth v0.16.1 // indirect
+	cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect
+	cloud.google.com/go/compute/metadata v0.6.0 // indirect
+	cloud.google.com/go/iam v1.5.2 // indirect
 	github.com/apache/arrow-go/v18 v18.1.0 // indirect
 	github.com/containerd/errdefs v1.0.0 // indirect
 	github.com/containerd/errdefs/pkg v0.3.0 // indirect
@@ -24,18 +31,26 @@ require (
 	github.com/go-viper/mapstructure/v2 v2.3.0 // indirect
 	github.com/goccy/go-json v0.10.5 // indirect
 	github.com/google/flatbuffers v25.1.24+incompatible // indirect
+	github.com/google/s2a-go v0.1.9 // indirect
+	github.com/googleapis/enterprise-certificate-proxy v0.3.6 // indirect
+	github.com/googleapis/gax-go/v2 v2.14.1 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.9 // indirect
 	github.com/moby/go-archive v0.1.0 // indirect
 	github.com/shirou/gopsutil/v4 v4.25.5 // indirect
 	github.com/zeebo/xxh3 v1.0.2 // indirect
+	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.60.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.35.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.6.0 // indirect
 	golang.org/x/exp v0.0.0-20250128182459-e0ece0dbea4c // indirect
 	golang.org/x/mod v0.22.0 // indirect
+	golang.org/x/oauth2 v0.30.0 // indirect
 	golang.org/x/time v0.11.0 // indirect
 	golang.org/x/tools v0.29.0 // indirect
 	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
+	google.golang.org/api v0.233.0 // indirect
+	google.golang.org/genproto v0.0.0-20250303144028-a0af3efb3deb // indirect
 )
 
 require (