diff --git a/internal/controller/health/grpc_handler.go b/internal/controller/health/grpc_handler.go new file mode 100644 index 0000000..7c7db11 --- /dev/null +++ b/internal/controller/health/grpc_handler.go @@ -0,0 +1,139 @@ +package health + +import ( + "context" + "errors" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/domain/activitypool" + healthDomain "proxy-pool/internal/domain/health" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +var ErrInvalidGRPCHandler = errors.New("invalid checker grpc handler") + +// CheckerIdentityAuthorizer proves that the calling mTLS identity owns the +// checker ID in the request. It is deliberately separate from worker session +// identity because Checkers do not own gateway snapshots. +type CheckerIdentityAuthorizer interface { + AuthorizeChecker(context.Context, string) error +} + +type GRPCHandlerOptions struct { + MaxObservationsPerBatch int +} + +func DefaultGRPCHandlerOptions() GRPCHandlerOptions { + return GRPCHandlerOptions{MaxObservationsPerBatch: 1_000} +} + +// GRPCHandler exposes fact reporting now. StreamCheckTasks remains inherited +// as unimplemented until the Controller's leased task scheduler is available. +type GRPCHandler struct { + controlplanev1.UnimplementedCheckerControlPlaneServer + reducer *Reducer + identity CheckerIdentityAuthorizer + options GRPCHandlerOptions +} + +func NewGRPCHandler( + reducer *Reducer, + identity CheckerIdentityAuthorizer, + options GRPCHandlerOptions, +) (*GRPCHandler, error) { + if reducer == nil || nilInterface(identity) { + return nil, ErrInvalidGRPCHandler + } + if options.MaxObservationsPerBatch == 0 { + options = DefaultGRPCHandlerOptions() + } + if options.MaxObservationsPerBatch < 0 { + return nil, ErrInvalidGRPCHandler + } + return &GRPCHandler{reducer: reducer, identity: identity, options: options}, nil +} + +func (handler *GRPCHandler) ReportObservations( + ctx context.Context, + request *controlplanev1.ObservationBatch, +) (*controlplanev1.ReportObservationsResponse, error) { + if ctx == nil || handler == nil || handler.reducer == nil || nilInterface(handler.identity) || request == nil || + request.GetCheckerId() == "" || len(request.GetObservations()) == 0 || + len(request.GetObservations()) > handler.options.MaxObservationsPerBatch { + return nil, healthGRPCError(ErrInvalidGRPCHandler) + } + if err := handler.identity.AuthorizeChecker(ctx, request.GetCheckerId()); err != nil { + return nil, status.Error(codes.PermissionDenied, "checker identity is not authorized") + } + response := &controlplanev1.ReportObservationsResponse{} + for _, item := range request.GetObservations() { + observation, err := decodeHealthObservation(item) + if err == nil { + _, err = handler.reducer.Apply(ctx, observation) + } + if err == nil { + response.Accepted++ + continue + } + if rejectedObservationError(err) { + response.Rejected++ + continue + } + return nil, healthGRPCError(err) + } + return response, nil +} + +func decodeHealthObservation(item *controlplanev1.HealthObservation) (healthDomain.Observation, error) { + if item == nil || item.GetLatency() == nil || item.GetLatency().CheckValid() != nil || + item.GetObservedAt() == nil || item.GetObservedAt().CheckValid() != nil { + return healthDomain.Observation{}, healthDomain.ErrInvalidObservation + } + level, ok := decodeCheckLevel(item.GetLevel()) + if !ok { + return healthDomain.Observation{}, healthDomain.ErrInvalidObservation + } + return healthDomain.Observation{ + TaskID: item.GetTaskId(), ProxyID: item.GetProxyId(), Level: level, + RoutingName: item.GetRoutingName(), TargetURL: item.GetTargetUrl(), Success: item.GetSuccess(), + FailureClass: item.GetFailureClass(), Latency: item.GetLatency().AsDuration(), + ObservedEgressIP: item.GetObservedEgressIp(), ObservedAt: item.GetObservedAt().AsTime(), + }, nil +} + +func decodeCheckLevel(level controlplanev1.CheckLevel) (healthDomain.Level, bool) { + switch level { + case controlplanev1.CheckLevel_CHECK_LEVEL_BASIC: + return healthDomain.LevelBasic, true + case controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS: + return healthDomain.LevelEgress, true + case controlplanev1.CheckLevel_CHECK_LEVEL_TARGET: + return healthDomain.LevelTarget, true + default: + return "", false + } +} + +func rejectedObservationError(err error) bool { + return errors.Is(err, healthDomain.ErrInvalidObservation) || errors.Is(err, healthDomain.ErrInvalidFailureThreshold) || + errors.Is(err, healthDomain.ErrStaleObservation) || errors.Is(err, healthDomain.ErrConflictingObservation) || + errors.Is(err, activitypool.ErrInvalidHealthUpdate) || errors.Is(err, activitypool.ErrActivityNotFound) || + errors.Is(err, ErrInvalidThreshold) +} + +func healthGRPCError(err error) error { + switch { + case errors.Is(err, context.Canceled): + return status.Error(codes.Canceled, "checker control request canceled") + case errors.Is(err, context.DeadlineExceeded): + return status.Error(codes.DeadlineExceeded, "checker control request deadline exceeded") + case errors.Is(err, ErrInvalidGRPCHandler), errors.Is(err, healthDomain.ErrInvalidObservation), + errors.Is(err, healthDomain.ErrInvalidFailureThreshold), errors.Is(err, activitypool.ErrInvalidHealthUpdate), + errors.Is(err, ErrInvalidThreshold): + return status.Error(codes.InvalidArgument, "invalid checker observation batch") + default: + return status.Error(codes.Unavailable, "checker control plane unavailable") + } +} diff --git a/internal/controller/health/grpc_handler_test.go b/internal/controller/health/grpc_handler_test.go new file mode 100644 index 0000000..86ad896 --- /dev/null +++ b/internal/controller/health/grpc_handler_test.go @@ -0,0 +1,119 @@ +package health + +import ( + "context" + "errors" + "testing" + "time" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + healthDomain "proxy-pool/internal/domain/health" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/durationpb" + "google.golang.org/protobuf/types/known/timestamppb" +) + +func TestGRPCHandlerReportsAcceptedAndRejectedObservations(t *testing.T) { + global := &recordingGlobalStore{} + target := &recordingTargetStore{} + reducer, err := NewReducer(global, target, func(context.Context, healthDomain.Observation) (int, error) { return 2, nil }) + if err != nil { + t.Fatalf("NewReducer(): %v", err) + } + identity := &recordingCheckerIdentity{} + handler, err := NewGRPCHandler(reducer, identity, GRPCHandlerOptions{MaxObservationsPerBatch: 3}) + if err != nil { + t.Fatalf("NewGRPCHandler(): %v", err) + } + now := time.Date(2026, 7, 31, 14, 0, 0, 0, time.UTC) + response, err := handler.ReportObservations(context.Background(), &controlplanev1.ObservationBatch{ + CheckerId: "checker-a", + Observations: []*controlplanev1.HealthObservation{ + grpcHealthObservation("task-basic", controlplanev1.CheckLevel_CHECK_LEVEL_BASIC, now), + grpcHealthObservation("task-target", controlplanev1.CheckLevel_CHECK_LEVEL_TARGET, now.Add(time.Second)), + {TaskId: "invalid", ProxyId: "proxy-a", Level: controlplanev1.CheckLevel_CHECK_LEVEL_UNSPECIFIED}, + }, + }) + if err != nil || response.GetAccepted() != 2 || response.GetRejected() != 1 { + t.Fatalf("ReportObservations() = (%+v, %v)", response, err) + } + if identity.checkerID != "checker-a" || len(global.commands) != 1 || len(target.commands) != 1 { + t.Fatalf("identity=%q global=%d target=%d", identity.checkerID, len(global.commands), len(target.commands)) + } +} + +func TestGRPCHandlerRejectsInvalidCallsBeforeStoreMutation(t *testing.T) { + global := &recordingGlobalStore{} + target := &recordingTargetStore{} + reducer, err := NewReducer(global, target, func(context.Context, healthDomain.Observation) (int, error) { return 2, nil }) + if err != nil { + t.Fatalf("NewReducer(): %v", err) + } + identity := &recordingCheckerIdentity{err: errors.New("not allowed")} + handler, err := NewGRPCHandler(reducer, identity, GRPCHandlerOptions{MaxObservationsPerBatch: 1}) + if err != nil { + t.Fatalf("NewGRPCHandler(): %v", err) + } + _, err = handler.ReportObservations(context.Background(), &controlplanev1.ObservationBatch{ + CheckerId: "checker-a", Observations: []*controlplanev1.HealthObservation{ + grpcHealthObservation("task-basic", controlplanev1.CheckLevel_CHECK_LEVEL_BASIC, time.Now()), + }, + }) + if status.Code(err) != codes.PermissionDenied || len(global.commands) != 0 || len(target.commands) != 0 { + t.Fatalf("ReportObservations(denied) = %v; global=%d target=%d", err, len(global.commands), len(target.commands)) + } + _, err = handler.ReportObservations(context.Background(), &controlplanev1.ObservationBatch{ + CheckerId: "checker-a", Observations: []*controlplanev1.HealthObservation{ + grpcHealthObservation("task-one", controlplanev1.CheckLevel_CHECK_LEVEL_BASIC, time.Now()), + grpcHealthObservation("task-two", controlplanev1.CheckLevel_CHECK_LEVEL_BASIC, time.Now()), + }, + }) + if status.Code(err) != codes.InvalidArgument { + t.Fatalf("ReportObservations(too large) = %v, want InvalidArgument", err) + } +} + +func TestGRPCHandlerReturnsUnavailableForStoreFailure(t *testing.T) { + global := &recordingGlobalStore{err: errors.New("redis unavailable")} + target := &recordingTargetStore{} + reducer, err := NewReducer(global, target, func(context.Context, healthDomain.Observation) (int, error) { return 2, nil }) + if err != nil { + t.Fatalf("NewReducer(): %v", err) + } + handler, err := NewGRPCHandler(reducer, &recordingCheckerIdentity{}, GRPCHandlerOptions{MaxObservationsPerBatch: 1}) + if err != nil { + t.Fatalf("NewGRPCHandler(): %v", err) + } + _, err = handler.ReportObservations(context.Background(), &controlplanev1.ObservationBatch{ + CheckerId: "checker-a", Observations: []*controlplanev1.HealthObservation{ + grpcHealthObservation("task-basic", controlplanev1.CheckLevel_CHECK_LEVEL_BASIC, time.Now()), + }, + }) + if status.Code(err) != codes.Unavailable { + t.Fatalf("ReportObservations(store failure) = %v, want Unavailable", err) + } +} + +func grpcHealthObservation(taskID string, level controlplanev1.CheckLevel, observedAt time.Time) *controlplanev1.HealthObservation { + item := &controlplanev1.HealthObservation{ + TaskId: taskID, ProxyId: "proxy-a", Level: level, Success: true, + Latency: durationpb.New(time.Millisecond), ObservedAt: timestamppb.New(observedAt), + } + if level == controlplanev1.CheckLevel_CHECK_LEVEL_TARGET { + item.RoutingName = "route-a" + item.TargetUrl = "https://target.example/check" + } + return item +} + +type recordingCheckerIdentity struct { + checkerID string + err error +} + +func (identity *recordingCheckerIdentity) AuthorizeChecker(_ context.Context, checkerID string) error { + identity.checkerID = checkerID + return identity.err +} diff --git a/internal/controller/health/reducer_test.go b/internal/controller/health/reducer_test.go index 4ac54d3..5a4843d 100644 --- a/internal/controller/health/reducer_test.go +++ b/internal/controller/health/reducer_test.go @@ -79,11 +79,12 @@ func reducerObservation(level healthDomain.Level, observedAt time.Time) healthDo type recordingGlobalStore struct { commands []activitypool.GlobalHealthCommand + err error } func (store *recordingGlobalStore) ApplyGlobalObservation(_ context.Context, command activitypool.GlobalHealthCommand) (activitypool.Entry, error) { store.commands = append(store.commands, command) - return activitypool.Entry{}, nil + return activitypool.Entry{}, store.err } type recordingTargetStore struct {