pgwatch is a PostgreSQL-specific monitoring tool that normally persists collected metrics in a PostgreSQL database. That default has limits: some deployments need metrics in an OLAP system or in a particular format. pgwatch's gRPC client covers those cases — a compatible gRPC server that writes wherever you like plugs in as a sink. What follows is a proof of concept in Go that receives measurements from pgwatch and writes them to ClickHouse.

The Receiver contract

The pgwatch gRPC client is defined in the api/pb/pgwatch.proto protobuf file, whose main Receiver service exposes three methods:

  1. UpdateMeasurements(MeasurementEnvelope) returns (Reply)
  2. SyncMetric(SyncReq) returns (Reply)
  3. DefineMetrics(google.protobuf.Struct) returns (Reply)

Implementing these with the signatures gRPC expects, registering the handlers and starting the server is all that is required. pgwatch is then pointed at the server's URI and hands over metrics as they are collected.

Preparing the environment

Clone pgwatch locally if you have not already:

1

$ git clone https://github.com/cybertec-postgresql/pgwatch.git && cd pgwatch

Generating Go code from the protobuf definition requires three tools:

  • protoc, the Protocol Buffers compiler
  • protoc-gen-go, the Go plugin for protoc
  • protoc-gen-go-grpc, the Go gRPC plugin for protoc

1

2

3

4

5

6

7

8

9

10

11

# On Windows:

go install google.golang.org/protobuf/cmd/protoc-gen-go@latest

go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

winget install protobuf

# On Linux/macOS:

go install google.golang.org/protobuf/cmd/protoc-gen-go@latest

go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

# Install protoc via package manager (apt, brew, etc.)

Then run:

1

$ go generate ./api/pb

Create a directory for the server and copy the generated gRPC service code into it:

1

$ mkdir ~/pgwatch-server && cp -r ./api/pb ~/pgwatch-server/pb  && cd ~/pgwatch-server && go mod init pgwatch-server && go mod tidy

A ClickHouse-backed receiver

With the generated code in place, the first piece of server logic uses the clickhouse-go package to initialize a connection from the supplied credentials and create a Measurements table.

Note: The PoC keeps things as simple as possible by ignoring errors and storing every metric in a single table.

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

// ~/pgwatch-server/main.go

type ClickHouseReceiver struct {

    Conn driver.Conn

    pb.UnimplementedReceiverServer

}

func GetConnection(Username, Password, DBName, ServerURI string) driver.Conn {

    options := &clickhouse.Options{

        Addr:     []string{ServerURI},

        Protocol: clickhouse.Native,

        Auth: clickhouse.Auth{

            Database: DBName,

            Username: Username,

            Password: Password,

        },

    }

    conn, _ := clickhouse.Open(options)

    return conn

}

func NewClickHouseReceiver(User, Password, DBName, serverURI string) *ClickHouseReceiver {

    conn := GetConnection(User, Password, DBName, serverURI)

    chr := &ClickHouseReceiver{Conn: conn}

    chr.CreateMeasurementsTable()

    return chr

}

func (r *ClickHouseReceiver) CreateMeasurementsTable() {

    query :=

    `CREATE TABLE IF NOT EXISTS

    Measurements(dbname String, metric_name String, custom_tags Map(String, String), data String)

    ENGINE=MergeTree() ORDER BY tuple()`

    r.Conn.Exec(context.TODO(), query)

}

Next comes a custom UpdateMeasurements() that takes the data delivered by pgwatch's gRPC client and inserts it into ClickHouse.

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

// ~/pgwatch-server/main.go

func (r *ClickHouseReceiver) UpdateMeasurements(ctx context.Context, msg *pb.MeasurementEnvelope) (*pb.Reply, error) {

    batch, _ := r.Conn.PrepareBatch(ctx, `INSERT INTO Measurements (dbname, metric_name, custom_tags, data) VALUES (?, ?, ?, ?)`)

    for _, measurement := range msg.GetData() {

        measurementJson, _ := json.Marshal(measurement)

        batch.Append(

            msg.GetDBName(),

            msg.GetMetricName(),

            msg.GetCustomTags(),

            measurementJson,

        )

    }

    batch.Send()

    return &pb.Reply{Logmsg: "Measurement Written."}, nil

}

The remaining piece is the main function, which registers the ClickHouseReceiver struct as the handler and starts the server.

1

2

3

4

5

6

7

8

9

10

// ~/pgwatch-server/main.go

func main() {

    receiver := NewClickHouseReceiver("default", "", "mydb", "localhost:9000")

    lis, _ :=   net.Listen("tcp", "0.0.0.0:8888")  

     server := grpc.NewServer()

        pb.RegisterReceiverServer(server, receiver)  

     if err := server.Serve(lis); err != nil {

        log.Fatal(err)

   }

}

Testing it:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

$ go run main.go &

...

$ cd ~/pgwatch && go run ./cmd/pgwatch --sources=... --sink=grpc://localhost:8888

2025-11-18 14:20:34.553 [ERROR] [source:pgwatch] rpc error: code = Unimplemented desc = method SyncMetric not implemented

2025-11-18 14:20:34.600 [INFO] [source:pgwatch] [metric:psutil_disk_io_total] [rows:1] [cache:false] measurements fetched

2025-11-18 14:20:34.600 [INFO] [source:pgwatch] [metric:psutil_mem] [rows:1] [cache:false] measurements fetched

2025-11-18 14:20:34.603 [INFO] [sink:grpc] [address:localhost:8888] [rows:1] [elapsed:2.626398ms] measurements written

2025-11-18 14:20:34.603 [INFO] [sink:grpc] [address:localhost:8888] Measurement Written.

2025-11-18 14:20:34.605 [INFO] [sink:grpc] [address:localhost:8888] [rows:1] [elapsed:2.061434ms] measurements written

2025-11-18 14:20:34.605 [INFO] [sink:grpc] [address:localhost:8888] Measurement Written.

2025-11-18 14:20:34.618 [INFO] [source:pgwatch] [metric:psutil_disk] [rows:1] [cache:false] measurements fetched

2025-11-18 14:20:34.621 [INFO] [sink:grpc] [address:localhost:8888] [rows:1] [elapsed:2.495912ms] measurements written

2025-11-18 14:20:34.621 [INFO] [sink:grpc] [address:localhost:8888] Measurement Written

The logs show [address:localhost:8888] Measurement Written, and querying the database confirms the data arrived:

1

2

3

4

5

6

7

8

9

10

11

12

SELECT

    dbname,

    metric_name

FROM Measurements

LIMIT 4

   ┌─dbname──┬─metric_name──────────┐

1. │ pgwatch │ psutil_disk_io_total    │

2. │ pgwatch │ psutil_mem              │

3. │ pgwatch │ psutil_disk             │

4. │ pgwatch │ psutil_cpu              │

   └────────┴─────────────────────┘

Beyond the proof of concept

  • The gRPC client keeps pgwatch extensible without either boxing users in or forcing the maintainers to carry tens of storage options in-core. The trade-off is some development work plus an extra gRPC hop.
  • This example is a basic PoC sized for a blog post. Production use needs considerably more at the database layer — a separate table per metric, cached metrics, and so on.
  • Implement SyncMetric() to pre-allocate or free resources when metrics are added to or deleted from a monitored source:

1

2

3

4

5

6

7

func (r *ClickHouseReceiver) SyncMetric(ctx context.Context, req *pb.SyncReq) (*pb.Reply, error) {

    if req.GetOperation() == pb.SyncOp_AddOp {

        // take some action

    } else if req.GetOperation() == pb.SyncOp_DeleteOp {

        // take another action

    }

}
  • Implement DefineMetrics() to receive the definition of active metrics and presets:

1

2

3

func (r *ClickHouseReceiver) DefineMetrics(ctx context.Context, st *structpb.Struct) (*pb.Reply, error) {

    // ...

}
  • The pgwatch-contrib repository holds community-maintained gRPC server implementations worth studying or extending.