Add health endpoint for worker (#2442)

* Return worker process information in the health endpoint.

Experimental:  The worker_info and returned worker_process_info fields are experimental and can change or be removed without notice.

When a user passes in "worker_info=1" or "worker_info=true" and a worker is running the returned json will be returned with a key named "worker_process_info" which value is a json object with information about the worker process.

internal/daemon/controller/handler.go

@@ -39,7 +39,7 @@ import (
 	"github.com/hashicorp/boundary/internal/daemon/controller/internal/metric"
 	"github.com/hashicorp/boundary/internal/gen/controller/api/services"
 	authpb "github.com/hashicorp/boundary/internal/gen/controller/auth"
-	opsservices "github.com/hashicorp/boundary/internal/gen/controller/ops/services"
+	opsservices "github.com/hashicorp/boundary/internal/gen/ops/services"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/go-cleanhttp"
 	"github.com/hashicorp/go-secure-stdlib/listenerutil"

internal/daemon/controller/handlers/health/health.go

@@ -2,16 +2,22 @@ package health
 
 import (
 	"context"
+	"fmt"
 	"net/http"
+	"sync"
 
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers"
-	pbs "github.com/hashicorp/boundary/internal/gen/controller/ops/services"
+	pbs "github.com/hashicorp/boundary/internal/gen/ops/services"
+	pbhealth "github.com/hashicorp/boundary/internal/gen/worker/health"
 )
 
 type Service struct {
 	pbs.UnsafeHealthServiceServer
-
 	replyWithServiceUnavailable bool
+
+	workerInfoLock sync.RWMutex
+	workerInfoOnce sync.Once
+	workerInfoFn   func() *pbhealth.HealthInfo
 }
 
 var _ pbs.HealthServiceServer = (*Service)(nil)
@@ -22,15 +28,25 @@ func NewService() *Service {
 }
 
 func (s *Service) GetHealth(ctx context.Context, req *pbs.GetHealthRequest) (*pbs.GetHealthResponse, error) {
+	resp := &pbs.GetHealthResponse{}
+
+	if req.GetWorkerInfo() {
+		func() {
+			s.workerInfoLock.RLock()
+			defer s.workerInfoLock.RUnlock()
+			if s.workerInfoFn != nil {
+				resp.WorkerProcessInfo = s.workerInfoFn()
+			}
+		}()
+	}
 	if s.replyWithServiceUnavailable {
 		err := handlers.SetStatusCode(ctx, http.StatusServiceUnavailable)
 		if err != nil {
 			return nil, err
 		}
-		return &pbs.GetHealthResponse{}, nil
 	}
 
-	return &pbs.GetHealthResponse{}, nil
+	return resp, nil
 }
 
 // StartServiceUnavailableReplies gets returned to the caller of NewService.
@@ -38,3 +54,17 @@ func (s *Service) GetHealth(ctx context.Context, req *pbs.GetHealthRequest) (*pb
 func (s *Service) StartServiceUnavailableReplies() {
 	s.replyWithServiceUnavailable = true
 }
+
+// SetWorkerProcessInformationFunc sets the function that can be used to get the
+// current worker related process information.
+func (s *Service) SetWorkerProcessInformationFunc(fn func() *pbhealth.HealthInfo) error {
+	if fn == nil {
+		return fmt.Errorf("CurrentWorkerConnection function was nil")
+	}
+	s.workerInfoLock.Lock()
+	defer s.workerInfoLock.Unlock()
+	s.workerInfoOnce.Do(func() {
+		s.workerInfoFn = fn
+	})
+	return nil
+}

internal/daemon/controller/handlers/health/health_test.go

@@ -5,17 +5,32 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/hashicorp/boundary/internal/gen/controller/ops/services"
+	"github.com/google/go-cmp/cmp"
+	"github.com/hashicorp/boundary/internal/gen/ops/services"
+	pbhealth "github.com/hashicorp/boundary/internal/gen/worker/health"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
+	"google.golang.org/protobuf/testing/protocmp"
+	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
 func TestGetHealth(t *testing.T) {
+	workerInfo := &pbhealth.HealthInfo{
+		ActiveSessionCount: wrapperspb.UInt32(2),
+		SessionConnections: map[string]uint32{
+			"foo": uint32(2),
+			"bar": uint32(1),
+		},
+	}
+
 	tests := []struct {
 		name                 string
 		ctx                  context.Context
+		workerInfoFn         func() *pbhealth.HealthInfo
 		serviceUnavailable   bool
+		request              *services.GetHealthRequest
 		expGetHealthResponse *services.GetHealthResponse
 		expErr               bool
 		expErrMsg            string
@@ -24,20 +39,50 @@ func TestGetHealth(t *testing.T) {
 			name:                 "healthy reply",
 			ctx:                  context.Background(),
 			serviceUnavailable:   false,
+			request:              &services.GetHealthRequest{},
 			expGetHealthResponse: &services.GetHealthResponse{},
 			expErr:               false,
 		},
+		{
+			name:                 "with worker info no query parameter",
+			ctx:                  context.Background(),
+			workerInfoFn:         func() *pbhealth.HealthInfo { return workerInfo },
+			serviceUnavailable:   false,
+			request:              &services.GetHealthRequest{},
+			expGetHealthResponse: &services.GetHealthResponse{},
+			expErr:               false,
+		},
+		{
+			name:                 "with worker info and query param",
+			ctx:                  context.Background(),
+			workerInfoFn:         func() *pbhealth.HealthInfo { return workerInfo },
+			serviceUnavailable:   false,
+			request:              &services.GetHealthRequest{WorkerInfo: true},
+			expGetHealthResponse: &services.GetHealthResponse{WorkerProcessInfo: workerInfo},
+			expErr:               false,
+		},
 		{
 			name:                 "service unavailable reply",
 			ctx:                  grpc.NewContextWithServerTransportStream(context.Background(), &testServerTransportStream{expHttpCode: "503"}),
 			serviceUnavailable:   true,
+			request:              &services.GetHealthRequest{},
 			expGetHealthResponse: &services.GetHealthResponse{},
 			expErr:               false,
 		},
+		{
+			name:                 "service unavailable with worker info reply",
+			ctx:                  grpc.NewContextWithServerTransportStream(context.Background(), &testServerTransportStream{expHttpCode: "503"}),
+			workerInfoFn:         func() *pbhealth.HealthInfo { return workerInfo },
+			serviceUnavailable:   true,
+			request:              &services.GetHealthRequest{WorkerInfo: true},
+			expGetHealthResponse: &services.GetHealthResponse{WorkerProcessInfo: workerInfo},
+			expErr:               false,
+		},
 		{
 			name:               "get health error",
 			ctx:                context.Background(),
 			serviceUnavailable: true,
+			request:            &services.GetHealthRequest{},
 			expErr:             true,
 			expErrMsg:          "handlers.SetHttpStatusCode: internal error, unknown: error #500: rpc error: code = Internal desc = grpc: failed to fetch the stream from the context context.Background",
 		},
@@ -47,11 +92,15 @@ func TestGetHealth(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			hs := NewService()
 
+			if tt.workerInfoFn != nil {
+				require.NoError(t, hs.SetWorkerProcessInformationFunc(tt.workerInfoFn))
+			}
+
 			if tt.serviceUnavailable {
 				hs.StartServiceUnavailableReplies()
 			}
 
-			rsp, err := hs.GetHealth(tt.ctx, &services.GetHealthRequest{})
+			rsp, err := hs.GetHealth(tt.ctx, tt.request)
 			if tt.expErr {
 				require.Error(t, err)
 				require.EqualError(t, err, tt.expErrMsg)
@@ -59,7 +108,7 @@ func TestGetHealth(t *testing.T) {
 			}
 
 			require.NoError(t, err)
-			require.EqualValues(t, tt.expGetHealthResponse, rsp)
+			assert.Empty(t, cmp.Diff(tt.expGetHealthResponse, rsp, protocmp.Transform()))
 		})
 	}
 }