[!] add RPC sink support, closes #464

src/sinks/multiwriter.go

@@ -43,6 +43,8 @@ func NewMultiWriter(ctx context.Context, opts *config.Options, metricDefs *metri
 			w, err = NewPostgresWriter(ctx, s, &opts.Measurements, metricDefs)
 		case "prometheus":
 			w, err = NewPrometheusWriter(ctx, path)
+		case "rpc":
+			w, err = NewRPCWriter(ctx, path)
 		default:
 			return nil, fmt.Errorf("unknown schema %s in sink URI %s", scheme, s)
 		}

src/sinks/rpc.go

@@ -0,0 +1,81 @@
+package sinks
+
+import (
+	"context"
+	"net/rpc"
+
+	"github.com/cybertec-postgresql/pgwatch3/log"
+	"github.com/cybertec-postgresql/pgwatch3/metrics"
+)
+
+type RPCWriter struct {
+	ctx     context.Context
+	address string
+	client  *rpc.Client
+}
+
+func NewRPCWriter(ctx context.Context, address string) (*RPCWriter, error) {
+	client, err := rpc.DialHTTP("tcp", address)
+	if err != nil {
+		return nil, err
+	}
+
+	rw := &RPCWriter{
+		ctx:     ctx,
+		address: address,
+		client:  client,
+	}
+	go rw.watchCtx()
+	return rw, nil
+}
+
+// Sends Measurement Message to RPC Sink
+func (rw *RPCWriter) Write(msgs []metrics.MeasurementMessage) error {
+	if rw.ctx.Err() != nil {
+		return rw.ctx.Err()
+	}
+	if len(msgs) == 0 {
+		return nil
+	}
+	l := log.GetLogger(rw.ctx).
+		WithField("sink", "rpc").
+		WithField("address", rw.address)
+	for _, msg := range msgs {
+		var logMsg string
+		if err := rw.client.Call("Receiver.UpdateMeasurements", &msg, &logMsg); err != nil {
+			return err
+		}
+		if len(logMsg) > 0 {
+			l.Info(logMsg)
+		}
+	}
+	return nil
+}
+
+type SyncReq struct {
+	DbName     string
+	MetricName string
+	Operation  string
+}
+
+func (rw *RPCWriter) SyncMetric(dbUnique string, metricName string, op string) error {
+	var logMsg string
+	if err := rw.client.Call("Receiver.SyncMetric", &SyncReq{
+		Operation:  op,
+		DbName:     dbUnique,
+		MetricName: metricName,
+	}, &logMsg); err != nil {
+		return err
+	}
+	if len(logMsg) > 0 {
+		log.GetLogger(rw.ctx).
+			WithField("sink", "rpc").
+			WithField("address", rw.address).Info(logMsg)
+	}
+	return nil
+}
+
+func (rw *RPCWriter) watchCtx() {
+	<-rw.ctx.Done()
+	rw.client.Close()
+}

src/sinks/rpc_test.go

@@ -0,0 +1,125 @@
+package sinks_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"net/rpc"
+	"testing"
+
+	"github.com/cybertec-postgresql/pgwatch3/metrics"
+	"github.com/cybertec-postgresql/pgwatch3/sinks"
+	"github.com/stretchr/testify/assert"
+)
+
+type Receiver struct {
+}
+
+var ctxt = context.Background()
+
+func (receiver *Receiver) UpdateMeasurements(msg *metrics.MeasurementMessage, logMsg *string) error {
+	if msg == nil {
+		return errors.New("msgs is nil")
+	}
+	if msg.DBName != "Db" {
+		return errors.New("invalid message")
+	}
+	*logMsg = fmt.Sprintf("Received: %+v", *msg)
+	return nil
+}
+
+func (receiver *Receiver) SyncMetric(syncReq *sinks.SyncReq, logMsg *string) error {
+	if syncReq == nil {
+		return errors.New("msgs is nil")
+	}
+	if syncReq.Operation == "invalid" {
+		return errors.New("invalid message")
+	}
+	*logMsg = fmt.Sprintf("Received: %+v", *syncReq)
+	return nil
+}
+
+func init() {
+	recv := new(Receiver)
+	if err := rpc.Register(recv); err != nil {
+		panic(err)
+	}
+	rpc.HandleHTTP()
+	if listener, err := net.Listen("tcp", "0.0.0.0:5050"); err == nil {
+		go func() {
+			_ = http.Serve(listener, nil)
+		}()
+	} else {
+		panic(err)
+	}
+}
+
+// Test begin from here ---------------------------------------------------------
+func TestNewRPCWriter(t *testing.T) {
+	a := assert.New(t)
+	_, err := sinks.NewRPCWriter(ctxt, "foo")
+	a.Error(err)
+}
+
+func TestRPCWrite(t *testing.T) {
+	a := assert.New(t)
+	rw, err := sinks.NewRPCWriter(ctxt, "0.0.0.0:5050")
+	a.NoError(err)
+
+	// no error for valid messages
+	msgs := []metrics.MeasurementMessage{
+		{
+			DBName: "Db",
+		},
+	}
+	err = rw.Write(msgs)
+	a.NoError(err)
+
+	// error for invalid messages
+	msgs = []metrics.MeasurementMessage{
+		{
+			DBName: "invalid",
+		},
+	}
+	err = rw.Write(msgs)
+	a.Error(err)
+
+	// no error for empty messages
+	err = rw.Write([]metrics.MeasurementMessage{})
+	a.NoError(err)
+
+	// error for cancelled context
+	ctx, cancel := context.WithCancel(ctxt)
+	rw, err = sinks.NewRPCWriter(ctx, "0.0.0.0:5050")
+	a.NoError(err)
+	cancel()
+	err = rw.Write(msgs)
+	a.Error(err)
+}
+
+func TestRPCSyncMetric(t *testing.T) {
+	port := 5050
+	a := assert.New(t)
+	rw, err := sinks.NewRPCWriter(ctxt, "0.0.0.0:"+fmt.Sprint(port))
+	if err != nil {
+		t.Error("Unable to send sync metric signal")
+	}
+
+	// no error for valid messages
+	err = rw.SyncMetric("Test-DB", "DB-Metric", "Add")
+	a.NoError(err)
+
+	// error for invalid messages
+	err = rw.SyncMetric("", "", "invalid")
+	a.Error(err)
+
+	// error for cancelled context
+	ctx, cancel := context.WithCancel(ctxt)
+	rw, err = sinks.NewRPCWriter(ctx, "0.0.0.0:5050")
+	a.NoError(err)
+	cancel()
+	err = rw.SyncMetric("Test-DB", "DB-Metric", "Add")
+	a.Error(err)
+}