connectrpc health api

Signed-off-by: Joshua Kim <[email protected]>

Author: mpignatelli12

api/health/connecthandler/health_connect.go

@@ -0,0 +1,129 @@
+// Copyright (C) 2019-2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package connecthandler
+
+import (
+	"context"
+	"time"
+
+	"connectrpc.com/connect"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"github.com/ava-labs/avalanchego/api/health"
+	healthv1 "github.com/ava-labs/avalanchego/proto/pb/health/v1"
+)
+
+// NewConnectHealthService returns a ConnectRPC-compatible ConnectHealthService
+// that delegates calls to the existing health service implementation.
+func NewConnectHealthService(healthService health.Health) *ConnectHealthService {
+	return &ConnectHealthService{
+		Service: healthService,
+	}
+}
+
+type ConnectHealthService struct {
+	Service health.Health
+}
+
+func (s *ConnectHealthService) Readiness(
+	_ context.Context,
+	req *connect.Request[healthv1.APIArgs],
+) (*connect.Response[healthv1.APIReply], error) {
+	// The health.Health interface has different methods than the service.
+	// We need to use the methods of the Health interface.
+	checks, healthy := s.Service.Readiness(req.Msg.Tags...)
+
+	// Convert the health.Result map to protobuf format
+	checksProto := make(map[string]*healthv1.Result, len(checks))
+	for name, check := range checks {
+		checksProto[name] = convertResult(check)
+	}
+
+	out := &healthv1.APIReply{
+		Checks:  checksProto,
+		Healthy: healthy,
+	}
+
+	return connect.NewResponse(out), nil
+}
+
+func (s *ConnectHealthService) Health(
+	_ context.Context,
+	req *connect.Request[healthv1.APIArgs],
+) (*connect.Response[healthv1.APIReply], error) {
+	// The health.Health interface has different methods than the service.
+	// We need to use the methods of the Health interface.
+	checks, healthy := s.Service.Health(req.Msg.Tags...)
+
+	// Convert the health.Result map to protobuf format
+	checksProto := make(map[string]*healthv1.Result, len(checks))
+	for name, check := range checks {
+		checksProto[name] = convertResult(check)
+	}
+
+	out := &healthv1.APIReply{
+		Checks:  checksProto,
+		Healthy: healthy,
+	}
+
+	return connect.NewResponse(out), nil
+}
+
+func (s *ConnectHealthService) Liveness(
+	_ context.Context,
+	req *connect.Request[healthv1.APIArgs],
+) (*connect.Response[healthv1.APIReply], error) {
+	// The health.Health interface has different methods than the service.
+	// We need to use the methods of the Health interface.
+	checks, healthy := s.Service.Liveness(req.Msg.Tags...)
+
+	// Convert the health.Result map to protobuf format
+	checksProto := make(map[string]*healthv1.Result, len(checks))
+	for name, check := range checks {
+		checksProto[name] = convertResult(check)
+	}
+
+	out := &healthv1.APIReply{
+		Checks:  checksProto,
+		Healthy: healthy,
+	}
+
+	return connect.NewResponse(out), nil
+}
+
+// convertResult transforms a health.Result into a healthv1.Result
+func convertResult(r health.Result) *healthv1.Result {
+	result := &healthv1.Result{
+		Message:            "", // Will be set below if details exist
+		ContiguousFailures: r.ContiguousFailures,
+		DurationNs:         r.Duration.Nanoseconds(),
+	}
+
+	// Handle message field (from the Details field)
+	if r.Details != nil {
+		if msg, ok := r.Details.(string); ok {
+			result.Message = msg
+		}
+	}
+
+	// Set error field if exists
+	if r.Error != nil {
+		result.Error = *r.Error
+	}
+
+	// Set timestamp if not zero
+	if !r.Timestamp.IsZero() {
+		result.Timestamp = timestamppb.New(r.Timestamp)
+	}
+
+	// Set time of first failure if exists and not nil
+	if r.TimeOfFirstFailure != nil {
+		result.TimeOfFirstFailure = timestamppb.New(*r.TimeOfFirstFailure)
+	} else {
+		// Use zero time if TimeOfFirstFailure is nil
+		result.TimeOfFirstFailure = timestamppb.New(time.Time{})
+	}
+
+	return result
+}

api/info/connect_handler/connect_handler.go renamed to api/info/connecthandler/info_connect.go

@@ -4,12 +4,17 @@
 package node
 
 import (
+	"connectrpc.com/grpcreflect"
 	"context"
 	"crypto"
 	"crypto/tls"
 	"encoding/json"
 	"errors"
 	"fmt"
+	healthhandler "github.com/ava-labs/avalanchego/api/health/connecthandler"
+	infohandler "github.com/ava-labs/avalanchego/api/info/connecthandler"
+	"github.com/ava-labs/avalanchego/proto/pb/health/v1/healthv1connect"
+	"github.com/ava-labs/avalanchego/proto/pb/info/v1/infov1connect"
 	"io"
 	"io/fs"
 	"net"
@@ -21,7 +26,6 @@ import (
 	"sync"
 	"time"
 
-	"connectrpc.com/grpcreflect"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -48,7 +52,6 @@ import (
 	"github.com/ava-labs/avalanchego/network/dialer"
 	"github.com/ava-labs/avalanchego/network/peer"
 	"github.com/ava-labs/avalanchego/network/throttling"
-	"github.com/ava-labs/avalanchego/proto/pb/info/v1/infov1connect"
 	"github.com/ava-labs/avalanchego/snow"
 	"github.com/ava-labs/avalanchego/snow/networking/benchlist"
 	"github.com/ava-labs/avalanchego/snow/networking/router"
@@ -80,7 +83,6 @@ import (
 	"github.com/ava-labs/avalanchego/vms/registry"
 	"github.com/ava-labs/avalanchego/vms/rpcchainvm/runtime"
 
-	connecthandler "github.com/ava-labs/avalanchego/api/info/connect_handler"
 	databasefactory "github.com/ava-labs/avalanchego/database/factory"
 	avmconfig "github.com/ava-labs/avalanchego/vms/avm/config"
 	platformconfig "github.com/ava-labs/avalanchego/vms/platformvm/config"
@@ -266,6 +268,26 @@ func New(
 		return nil, fmt.Errorf("couldn't initialize indexer: %w", err)
 	}
 
+	// mount InfoService onto the shared gRPC mux
+	infoPattern, infoHandler := infov1connect.NewInfoServiceHandler(
+		infohandler.NewConnectInfoService(n.info),
+	)
+	n.grpcMux.Handle(infoPattern, infoHandler)
+
+	// mount HealthService onto the same mux
+	healthPattern, healthHandler := healthv1connect.NewHealthServiceHandler(
+		healthhandler.NewConnectHealthService(n.health),
+	)
+	n.grpcMux.Handle(healthPattern, healthHandler)
+
+	// register reflection for both services, once
+	reflector := grpcreflect.NewStaticReflector(
+		infov1connect.InfoServiceName,
+		healthv1connect.HealthServiceName,
+	)
+	reflectPattern, reflectHandler := grpcreflect.NewHandlerV1(reflector)
+	n.grpcMux.Handle(reflectPattern, reflectHandler)
+
 	n.health.Start(context.TODO(), n.Config.HealthCheckFreq)
 	n.initProfiler()
 
@@ -310,6 +332,9 @@ type Node struct {
 	// Monitors node health and runs health checks
 	health health.Health
 
+	// Info service instance
+	info *info.Info
+
 	// Build and parse messages, for both network layer and chain manager
 	msgCreator message.Creator
 
@@ -352,6 +377,9 @@ type Node struct {
 	// Handles HTTP API calls
 	APIServer server.Server
 
+	// Shared HTTP/2 mux for ConnectRPC services
+	grpcMux *http.ServeMux
+
 	// This node's configuration
 	Config *node.Config
 
@@ -1010,6 +1038,10 @@ func (n *Node) initAPIServer() error {
 		n.Config.HTTPConfig.HTTPConfig,
 		n.Config.HTTPAllowedHosts,
 	)
+
+	// Set up a shared HTTP/2 mux for all ConnectRPC services
+	n.grpcMux = http.NewServeMux()
+	n.APIServer.AddHTTP2Handler(n.grpcMux)
 	return err
 }
 
@@ -1353,7 +1385,7 @@ func (n *Node) initInfoAPI() error {
 		return fmt.Errorf("problem creating proof of possession: %w", err)
 	}
 
-	service, info, err := info.NewService(
+	service, infoInst, err := info.NewService(
 		info.Parameters{
 			Version:   version.CurrentApp,
 			NodeID:    n.ID,
@@ -1376,29 +1408,12 @@ func (n *Node) initInfoAPI() error {
 	if err != nil {
 		return err
 	}
-
-	// Register the InfoService handler and gRPC reflection handler
-	infoPattern, infoHandler := infov1connect.NewInfoServiceHandler(connecthandler.NewConnectInfoService(info))
-
-	// Register the gRPC reflection handler for InfoService
-	refPattern, refHandler := grpcreflect.NewHandlerV1(
-		grpcreflect.NewStaticReflector(infov1connect.InfoServiceName),
-	)
-
-	// Create a new ServeMux to handle the InfoService and reflection handlers
-	mux := http.NewServeMux()
-	mux.Handle(infoPattern, infoHandler)
-	mux.Handle(refPattern, refHandler)
-
-	if !n.APIServer.AddHeaderRoute("info", mux) {
-		// TODO do not panic
-		panic("could not add info route")
-	}
+	n.info = infoInst
 
 	return n.APIServer.AddRoute(
 		service,
 		"info",
-		"info",
+		"",
 	)
 }
 
@@ -1413,10 +1428,13 @@ func (n *Node) initHealthAPI() error {
 		return err
 	}
 
-	n.health, err = health.New(n.Log, healthReg)
+	// Create the health service
+	healthService, err := health.New(n.Log, healthReg)
 	if err != nil {
 		return err
 	}
+	// Store the health service in the node's health field
+	n.health = healthService
 
 	if !n.Config.HealthAPIEnabled {
 		n.Log.Info("skipping health API initialization because it has been disabled")

node/node.go

@@ -0,0 +1,41 @@
+syntax = "proto3";
+package health.v1;
+
+import "google/protobuf/timestamp.proto";
+
+option go_package = "github.com/ava-labs/avalanchego/proto/pb/health/v1;healthv1";
+
+// HealthService is the API service for health checks on a node
+service HealthService {
+  rpc Readiness(APIArgs) returns (APIReply);
+  rpc Liveness(APIArgs) returns (APIReply);
+  rpc Health(APIArgs) returns (APIReply);
+}
+
+// Result is the result of a health check.
+message Result {
+  string message = 1;
+  // Error is the string representation of the error returned by the failing
+  // HealthCheck. The value is nil if the check passed.
+  string error = 2;
+  // Timestamp of the last HealthCheck.
+  google.protobuf.Timestamp timestamp = 3;
+  // Duration is the amount of time this HealthCheck last took to evaluate.
+  int64 duration_ns = 4;
+  // ContiguousFailures the HealthCheck has returned.
+  int64 contiguous_failures = 5;
+  // TimeOfFirstFailure of the HealthCheck,
+  google.protobuf.Timestamp time_of_first_failure = 6;
+}
+
+// APIReply is the response for Readiness, Health, and Liveness.
+message APIReply {
+  map<string, Result> checks = 1;
+  bool healthy = 2;
+}
+
+// APIArgs is the arguments for Readiness, Health, and Liveness.
+message APIArgs {
+  repeated string tags = 1;
+}
+