HTTP validator API: health endpoints (#13149)

* updating health endpoints

* updating tests

* updating tests

* moving where the header is written and adding allow origin header

* removing header

* Update validator/rpc/handlers_health.go

Co-authored-by: Radosław Kapka <rkapka@wp.pl>

* Update validator/rpc/handlers_health.go

Co-authored-by: Radosław Kapka <rkapka@wp.pl>

* Update validator/rpc/handlers_health.go

Co-authored-by: Radosław Kapka <rkapka@wp.pl>

* radek's comments

* Update handlers_health.go

Co-authored-by: Radosław Kapka <rkapka@wp.pl>

* adding the correct errors to handle error

---------

Co-authored-by: Radosław Kapka <rkapka@wp.pl>

io/logs/mock/BUILD.bazel

@@ -0,0 +1,9 @@
+load("@prysm//tools/go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["mock_stream.go"],
+    importpath = "github.com/prysmaticlabs/prysm/v4/io/logs/mock",
+    visibility = ["//visibility:public"],
+    deps = ["//async/event:go_default_library"],
+)

io/logs/mock/mock_stream.go

@@ -0,0 +1,27 @@
+package mock
+
+import "github.com/prysmaticlabs/prysm/v4/async/event"
+
+type MockStreamer struct {
+	logs [][]byte
+	feed *event.Feed
+}
+
+// NewMockStreamer creates a new instance of MockStreamer.
+// It's useful to set up the default state for the mock, like initializing the feed.
+func NewMockStreamer(logs [][]byte) *MockStreamer {
+	return &MockStreamer{
+		logs: logs,
+		feed: new(event.Feed),
+	}
+}
+
+// GetLastFewLogs returns the predefined logs.
+func (m *MockStreamer) GetLastFewLogs() [][]byte {
+	return m.logs
+}
+
+// LogsFeed returns the predefined event feed.
+func (m *MockStreamer) LogsFeed() *event.Feed {
+	return m.feed
+}

proto/prysm/v1alpha1/health.pb.go

@@ -27,7 +27,6 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
-// Deprecated: Marked as deprecated in proto/prysm/v1alpha1/health.proto.
 type LogsResponse struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -85,28 +84,28 @@ var file_proto_prysm_v1alpha1_health_proto_rawDesc = []byte{
 	0x6c, 0x65, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x61, 0x6e, 0x6e, 0x6f, 0x74, 0x61, 0x74, 0x69, 0x6f,
 	0x6e, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x1b, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65,
 	0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x65, 0x6d, 0x70, 0x74, 0x79, 0x2e,
-	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x26, 0x0a, 0x0c, 0x4c, 0x6f, 0x67, 0x73, 0x52, 0x65, 0x73,
+	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x22, 0x0a, 0x0c, 0x4c, 0x6f, 0x67, 0x73, 0x52, 0x65, 0x73,
 	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x6c, 0x6f, 0x67, 0x73, 0x18, 0x01, 0x20,
-	0x03, 0x28, 0x09, 0x52, 0x04, 0x6c, 0x6f, 0x67, 0x73, 0x3a, 0x02, 0x18, 0x01, 0x32, 0x88, 0x01,
+	0x03, 0x28, 0x09, 0x52, 0x04, 0x6c, 0x6f, 0x67, 0x73, 0x32, 0x85, 0x01, 0x0a, 0x06, 0x48, 0x65,
-	0x0a, 0x06, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x12, 0x7e, 0x0a, 0x10, 0x53, 0x74, 0x72, 0x65,
+	0x61, 0x6c, 0x74, 0x68, 0x12, 0x7b, 0x0a, 0x10, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x42, 0x65,
-	0x61, 0x6d, 0x42, 0x65, 0x61, 0x63, 0x6f, 0x6e, 0x4c, 0x6f, 0x67, 0x73, 0x12, 0x16, 0x2e, 0x67,
+	0x61, 0x63, 0x6f, 0x6e, 0x4c, 0x6f, 0x67, 0x73, 0x12, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c,
-	0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45,
+	0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79,
-	0x6d, 0x70, 0x74, 0x79, 0x1a, 0x23, 0x2e, 0x65, 0x74, 0x68, 0x65, 0x72, 0x65, 0x75, 0x6d, 0x2e,
+	0x1a, 0x23, 0x2e, 0x65, 0x74, 0x68, 0x65, 0x72, 0x65, 0x75, 0x6d, 0x2e, 0x65, 0x74, 0x68, 0x2e,
-	0x65, 0x74, 0x68, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x2e, 0x4c, 0x6f, 0x67,
+	0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x2e, 0x4c, 0x6f, 0x67, 0x73, 0x52, 0x65, 0x73,
-	0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x2b, 0x82, 0xd3, 0xe4, 0x93, 0x02,
+	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x28, 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x22, 0x12, 0x20, 0x2f,
-	0x22, 0x12, 0x20, 0x2f, 0x65, 0x74, 0x68, 0x2f, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31,
+	0x65, 0x74, 0x68, 0x2f, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x2f, 0x68, 0x65, 0x61,
-	0x2f, 0x68, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x2f, 0x6c, 0x6f, 0x67, 0x73, 0x2f, 0x73, 0x74, 0x72,
+	0x6c, 0x74, 0x68, 0x2f, 0x6c, 0x6f, 0x67, 0x73, 0x2f, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x30,
-	0x65, 0x61, 0x6d, 0x88, 0x02, 0x01, 0x30, 0x01, 0x42, 0x96, 0x01, 0x0a, 0x19, 0x6f, 0x72, 0x67,
+	0x01, 0x42, 0x96, 0x01, 0x0a, 0x19, 0x6f, 0x72, 0x67, 0x2e, 0x65, 0x74, 0x68, 0x65, 0x72, 0x65,
-	0x2e, 0x65, 0x74, 0x68, 0x65, 0x72, 0x65, 0x75, 0x6d, 0x2e, 0x65, 0x74, 0x68, 0x2e, 0x76, 0x31,
+	0x75, 0x6d, 0x2e, 0x65, 0x74, 0x68, 0x2e, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x42,
-	0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x42, 0x0b, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x50, 0x72,
+	0x0b, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x3a,
-	0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x3a, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f,
+	0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x70, 0x72, 0x79, 0x73, 0x6d,
-	0x6d, 0x2f, 0x70, 0x72, 0x79, 0x73, 0x6d, 0x61, 0x74, 0x69, 0x63, 0x6c, 0x61, 0x62, 0x73, 0x2f,
+	0x61, 0x74, 0x69, 0x63, 0x6c, 0x61, 0x62, 0x73, 0x2f, 0x70, 0x72, 0x79, 0x73, 0x6d, 0x2f, 0x76,
-	0x70, 0x72, 0x79, 0x73, 0x6d, 0x2f, 0x76, 0x34, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x70,
+	0x34, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x70, 0x72, 0x79, 0x73, 0x6d, 0x2f, 0x76, 0x31,
-	0x72, 0x79, 0x73, 0x6d, 0x2f, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x3b, 0x65, 0x74,
+	0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x3b, 0x65, 0x74, 0x68, 0xaa, 0x02, 0x15, 0x45, 0x74, 0x68,
-	0x68, 0xaa, 0x02, 0x15, 0x45, 0x74, 0x68, 0x65, 0x72, 0x65, 0x75, 0x6d, 0x2e, 0x45, 0x74, 0x68,
+	0x65, 0x72, 0x65, 0x75, 0x6d, 0x2e, 0x45, 0x74, 0x68, 0x2e, 0x56, 0x31, 0x61, 0x6c, 0x70, 0x68,
-	0x2e, 0x56, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0xca, 0x02, 0x15, 0x45, 0x74, 0x68, 0x65,
+	0x61, 0x31, 0xca, 0x02, 0x15, 0x45, 0x74, 0x68, 0x65, 0x72, 0x65, 0x75, 0x6d, 0x5c, 0x45, 0x74,
-	0x72, 0x65, 0x75, 0x6d, 0x5c, 0x45, 0x74, 0x68, 0x5c, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61,
+	0x68, 0x5c, 0x76, 0x31, 0x61, 0x6c, 0x70, 0x68, 0x61, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74,
-	0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x6f, 0x33,
 }
 
 var (
@@ -187,7 +186,6 @@ const _ = grpc.SupportPackageIsVersion6
 //
 // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
 type HealthClient interface {
-	// Deprecated: Do not use.
 	StreamBeaconLogs(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (Health_StreamBeaconLogsClient, error)
 }
 
@@ -199,7 +197,6 @@ func NewHealthClient(cc grpc.ClientConnInterface) HealthClient {
 	return &healthClient{cc}
 }
 
-// Deprecated: Do not use.
 func (c *healthClient) StreamBeaconLogs(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (Health_StreamBeaconLogsClient, error) {
 	stream, err := c.cc.NewStream(ctx, &_Health_serviceDesc.Streams[0], "/ethereum.eth.v1alpha1.Health/StreamBeaconLogs", opts...)
 	if err != nil {
@@ -234,7 +231,6 @@ func (x *healthStreamBeaconLogsClient) Recv() (*LogsResponse, error) {
 
 // HealthServer is the server API for Health service.
 type HealthServer interface {
-	// Deprecated: Do not use.
 	StreamBeaconLogs(*emptypb.Empty, Health_StreamBeaconLogsServer) error
 }
 

proto/prysm/v1alpha1/health.proto

@@ -16,18 +16,14 @@ option php_namespace = "Ethereum\\Eth\\v1alpha1";
 //
 // The health service is able to return important metadata about a beacon node
 // such being able to stream logs via gRPC.
-// DEPRECATED: This endpoint doesn't appear to be used and have been marked for deprecation.
 service Health {
     rpc StreamBeaconLogs(google.protobuf.Empty) returns (stream LogsResponse) {
-        option deprecated = true;
         option (google.api.http) = {
             get: "/eth/v1alpha1/health/logs/stream"
         };
     }
 }
 
-// DEPRECATED: StreamBeaconLogs endpoint doesn't appear to be used and have been marked for deprecation.
 message LogsResponse {
-  option deprecated = true;
   repeated string logs = 1;
 }

proto/prysm/v1alpha1/validator-client/web_api.pb.gw.go

@@ -510,94 +510,6 @@ func local_request_SlashingProtection_ImportSlashingProtection_0(ctx context.Con
 
 }
 
-func request_Health_GetBeaconNodeConnection_0(ctx context.Context, marshaler runtime.Marshaler, client HealthClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
-	var protoReq emptypb.Empty
-	var metadata runtime.ServerMetadata
-
-	msg, err := client.GetBeaconNodeConnection(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
-	return msg, metadata, err
-
-}
-
-func local_request_Health_GetBeaconNodeConnection_0(ctx context.Context, marshaler runtime.Marshaler, server HealthServer, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
-	var protoReq emptypb.Empty
-	var metadata runtime.ServerMetadata
-
-	msg, err := server.GetBeaconNodeConnection(ctx, &protoReq)
-	return msg, metadata, err
-
-}
-
-func request_Health_GetLogsEndpoints_0(ctx context.Context, marshaler runtime.Marshaler, client HealthClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
-	var protoReq emptypb.Empty
-	var metadata runtime.ServerMetadata
-
-	msg, err := client.GetLogsEndpoints(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
-	return msg, metadata, err
-
-}
-
-func local_request_Health_GetLogsEndpoints_0(ctx context.Context, marshaler runtime.Marshaler, server HealthServer, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
-	var protoReq emptypb.Empty
-	var metadata runtime.ServerMetadata
-
-	msg, err := server.GetLogsEndpoints(ctx, &protoReq)
-	return msg, metadata, err
-
-}
-
-func request_Health_GetVersion_0(ctx context.Context, marshaler runtime.Marshaler, client HealthClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
-	var protoReq emptypb.Empty
-	var metadata runtime.ServerMetadata
-
-	msg, err := client.GetVersion(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
-	return msg, metadata, err
-
-}
-
-func local_request_Health_GetVersion_0(ctx context.Context, marshaler runtime.Marshaler, server HealthServer, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
-	var protoReq emptypb.Empty
-	var metadata runtime.ServerMetadata
-
-	msg, err := server.GetVersion(ctx, &protoReq)
-	return msg, metadata, err
-
-}
-
-func request_Health_StreamBeaconLogs_0(ctx context.Context, marshaler runtime.Marshaler, client HealthClient, req *http.Request, pathParams map[string]string) (Health_StreamBeaconLogsClient, runtime.ServerMetadata, error) {
-	var protoReq emptypb.Empty
-	var metadata runtime.ServerMetadata
-
-	stream, err := client.StreamBeaconLogs(ctx, &protoReq)
-	if err != nil {
-		return nil, metadata, err
-	}
-	header, err := stream.Header()
-	if err != nil {
-		return nil, metadata, err
-	}
-	metadata.HeaderMD = header
-	return stream, metadata, nil
-
-}
-
-func request_Health_StreamValidatorLogs_0(ctx context.Context, marshaler runtime.Marshaler, client HealthClient, req *http.Request, pathParams map[string]string) (Health_StreamValidatorLogsClient, runtime.ServerMetadata, error) {
-	var protoReq emptypb.Empty
-	var metadata runtime.ServerMetadata
-
-	stream, err := client.StreamValidatorLogs(ctx, &protoReq)
-	if err != nil {
-		return nil, metadata, err
-	}
-	header, err := stream.Header()
-	if err != nil {
-		return nil, metadata, err
-	}
-	metadata.HeaderMD = header
-	return stream, metadata, nil
-
-}
-
 func request_Auth_Initialize_0(ctx context.Context, marshaler runtime.Marshaler, client AuthClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
 	var protoReq emptypb.Empty
 	var metadata runtime.ServerMetadata
@@ -1020,98 +932,6 @@ func RegisterSlashingProtectionHandlerServer(ctx context.Context, mux *runtime.S
 	return nil
 }
 
-// RegisterHealthHandlerServer registers the http handlers for service Health to "mux".
-// UnaryRPC     :call HealthServer directly.
-// StreamingRPC :currently unsupported pending https://github.com/grpc/grpc-go/issues/906.
-// Note that using this registration option will cause many gRPC library features to stop working. Consider using RegisterHealthHandlerFromEndpoint instead.
-func RegisterHealthHandlerServer(ctx context.Context, mux *runtime.ServeMux, server HealthServer) error {
-
-	mux.Handle("GET", pattern_Health_GetBeaconNodeConnection_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
-		ctx, cancel := context.WithCancel(req.Context())
-		defer cancel()
-		var stream runtime.ServerTransportStream
-		ctx = grpc.NewContextWithServerTransportStream(ctx, &stream)
-		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
-		rctx, err := runtime.AnnotateIncomingContext(ctx, mux, req, "/ethereum.validator.accounts.v2.Health/GetBeaconNodeConnection")
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-		resp, md, err := local_request_Health_GetBeaconNodeConnection_0(rctx, inboundMarshaler, server, req, pathParams)
-		md.HeaderMD, md.TrailerMD = metadata.Join(md.HeaderMD, stream.Header()), metadata.Join(md.TrailerMD, stream.Trailer())
-		ctx = runtime.NewServerMetadataContext(ctx, md)
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-
-		forward_Health_GetBeaconNodeConnection_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
-
-	})
-
-	mux.Handle("GET", pattern_Health_GetLogsEndpoints_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
-		ctx, cancel := context.WithCancel(req.Context())
-		defer cancel()
-		var stream runtime.ServerTransportStream
-		ctx = grpc.NewContextWithServerTransportStream(ctx, &stream)
-		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
-		rctx, err := runtime.AnnotateIncomingContext(ctx, mux, req, "/ethereum.validator.accounts.v2.Health/GetLogsEndpoints")
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-		resp, md, err := local_request_Health_GetLogsEndpoints_0(rctx, inboundMarshaler, server, req, pathParams)
-		md.HeaderMD, md.TrailerMD = metadata.Join(md.HeaderMD, stream.Header()), metadata.Join(md.TrailerMD, stream.Trailer())
-		ctx = runtime.NewServerMetadataContext(ctx, md)
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-
-		forward_Health_GetLogsEndpoints_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
-
-	})
-
-	mux.Handle("GET", pattern_Health_GetVersion_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
-		ctx, cancel := context.WithCancel(req.Context())
-		defer cancel()
-		var stream runtime.ServerTransportStream
-		ctx = grpc.NewContextWithServerTransportStream(ctx, &stream)
-		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
-		rctx, err := runtime.AnnotateIncomingContext(ctx, mux, req, "/ethereum.validator.accounts.v2.Health/GetVersion")
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-		resp, md, err := local_request_Health_GetVersion_0(rctx, inboundMarshaler, server, req, pathParams)
-		md.HeaderMD, md.TrailerMD = metadata.Join(md.HeaderMD, stream.Header()), metadata.Join(md.TrailerMD, stream.Trailer())
-		ctx = runtime.NewServerMetadataContext(ctx, md)
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-
-		forward_Health_GetVersion_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
-
-	})
-
-	mux.Handle("GET", pattern_Health_StreamBeaconLogs_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
-		err := status.Error(codes.Unimplemented, "streaming calls are not yet supported in the in-process transport")
-		_, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
-		runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-		return
-	})
-
-	mux.Handle("GET", pattern_Health_StreamValidatorLogs_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
-		err := status.Error(codes.Unimplemented, "streaming calls are not yet supported in the in-process transport")
-		_, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
-		runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-		return
-	})
-
-	return nil
-}
-
 // RegisterAuthHandlerServer registers the http handlers for service Auth to "mux".
 // UnaryRPC     :call AuthServer directly.
 // StreamingRPC :currently unsupported pending https://github.com/grpc/grpc-go/issues/906.
@@ -1708,171 +1528,6 @@ var (
 	forward_SlashingProtection_ImportSlashingProtection_0 = runtime.ForwardResponseMessage
 )
 
-// RegisterHealthHandlerFromEndpoint is same as RegisterHealthHandler but
-// automatically dials to "endpoint" and closes the connection when "ctx" gets done.
-func RegisterHealthHandlerFromEndpoint(ctx context.Context, mux *runtime.ServeMux, endpoint string, opts []grpc.DialOption) (err error) {
-	conn, err := grpc.Dial(endpoint, opts...)
-	if err != nil {
-		return err
-	}
-	defer func() {
-		if err != nil {
-			if cerr := conn.Close(); cerr != nil {
-				grpclog.Infof("Failed to close conn to %s: %v", endpoint, cerr)
-			}
-			return
-		}
-		go func() {
-			<-ctx.Done()
-			if cerr := conn.Close(); cerr != nil {
-				grpclog.Infof("Failed to close conn to %s: %v", endpoint, cerr)
-			}
-		}()
-	}()
-
-	return RegisterHealthHandler(ctx, mux, conn)
-}
-
-// RegisterHealthHandler registers the http handlers for service Health to "mux".
-// The handlers forward requests to the grpc endpoint over "conn".
-func RegisterHealthHandler(ctx context.Context, mux *runtime.ServeMux, conn *grpc.ClientConn) error {
-	return RegisterHealthHandlerClient(ctx, mux, NewHealthClient(conn))
-}
-
-// RegisterHealthHandlerClient registers the http handlers for service Health
-// to "mux". The handlers forward requests to the grpc endpoint over the given implementation of "HealthClient".
-// Note: the gRPC framework executes interceptors within the gRPC handler. If the passed in "HealthClient"
-// doesn't go through the normal gRPC flow (creating a gRPC client etc.) then it will be up to the passed in
-// "HealthClient" to call the correct interceptors.
-func RegisterHealthHandlerClient(ctx context.Context, mux *runtime.ServeMux, client HealthClient) error {
-
-	mux.Handle("GET", pattern_Health_GetBeaconNodeConnection_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
-		ctx, cancel := context.WithCancel(req.Context())
-		defer cancel()
-		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
-		rctx, err := runtime.AnnotateContext(ctx, mux, req, "/ethereum.validator.accounts.v2.Health/GetBeaconNodeConnection")
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-		resp, md, err := request_Health_GetBeaconNodeConnection_0(rctx, inboundMarshaler, client, req, pathParams)
-		ctx = runtime.NewServerMetadataContext(ctx, md)
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-
-		forward_Health_GetBeaconNodeConnection_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
-
-	})
-
-	mux.Handle("GET", pattern_Health_GetLogsEndpoints_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
-		ctx, cancel := context.WithCancel(req.Context())
-		defer cancel()
-		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
-		rctx, err := runtime.AnnotateContext(ctx, mux, req, "/ethereum.validator.accounts.v2.Health/GetLogsEndpoints")
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-		resp, md, err := request_Health_GetLogsEndpoints_0(rctx, inboundMarshaler, client, req, pathParams)
-		ctx = runtime.NewServerMetadataContext(ctx, md)
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-
-		forward_Health_GetLogsEndpoints_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
-
-	})
-
-	mux.Handle("GET", pattern_Health_GetVersion_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
-		ctx, cancel := context.WithCancel(req.Context())
-		defer cancel()
-		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
-		rctx, err := runtime.AnnotateContext(ctx, mux, req, "/ethereum.validator.accounts.v2.Health/GetVersion")
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-		resp, md, err := request_Health_GetVersion_0(rctx, inboundMarshaler, client, req, pathParams)
-		ctx = runtime.NewServerMetadataContext(ctx, md)
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-
-		forward_Health_GetVersion_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
-
-	})
-
-	mux.Handle("GET", pattern_Health_StreamBeaconLogs_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
-		ctx, cancel := context.WithCancel(req.Context())
-		defer cancel()
-		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
-		rctx, err := runtime.AnnotateContext(ctx, mux, req, "/ethereum.validator.accounts.v2.Health/StreamBeaconLogs")
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-		resp, md, err := request_Health_StreamBeaconLogs_0(rctx, inboundMarshaler, client, req, pathParams)
-		ctx = runtime.NewServerMetadataContext(ctx, md)
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-
-		forward_Health_StreamBeaconLogs_0(ctx, mux, outboundMarshaler, w, req, func() (proto.Message, error) { return resp.Recv() }, mux.GetForwardResponseOptions()...)
-
-	})
-
-	mux.Handle("GET", pattern_Health_StreamValidatorLogs_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
-		ctx, cancel := context.WithCancel(req.Context())
-		defer cancel()
-		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
-		rctx, err := runtime.AnnotateContext(ctx, mux, req, "/ethereum.validator.accounts.v2.Health/StreamValidatorLogs")
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-		resp, md, err := request_Health_StreamValidatorLogs_0(rctx, inboundMarshaler, client, req, pathParams)
-		ctx = runtime.NewServerMetadataContext(ctx, md)
-		if err != nil {
-			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
-			return
-		}
-
-		forward_Health_StreamValidatorLogs_0(ctx, mux, outboundMarshaler, w, req, func() (proto.Message, error) { return resp.Recv() }, mux.GetForwardResponseOptions()...)
-
-	})
-
-	return nil
-}
-
-var (
-	pattern_Health_GetBeaconNodeConnection_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2, 2, 3}, []string{"v2", "validator", "health", "node_connection"}, ""))
-
-	pattern_Health_GetLogsEndpoints_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2, 2, 3, 2, 4}, []string{"v2", "validator", "health", "logs", "endpoints"}, ""))
-
-	pattern_Health_GetVersion_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2, 2, 3}, []string{"v2", "validator", "health", "version"}, ""))
-
-	pattern_Health_StreamBeaconLogs_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2, 2, 3, 2, 4, 2, 5}, []string{"v2", "validator", "health", "logs", "beacon", "stream"}, ""))
-
-	pattern_Health_StreamValidatorLogs_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2, 2, 3, 2, 1, 2, 4}, []string{"v2", "validator", "health", "logs", "stream"}, ""))
-)
-
-var (
-	forward_Health_GetBeaconNodeConnection_0 = runtime.ForwardResponseMessage
-
-	forward_Health_GetLogsEndpoints_0 = runtime.ForwardResponseMessage
-
-	forward_Health_GetVersion_0 = runtime.ForwardResponseMessage
-
-	forward_Health_StreamBeaconLogs_0 = runtime.ForwardResponseStream
-
-	forward_Health_StreamValidatorLogs_0 = runtime.ForwardResponseStream
-)
-
 // RegisterAuthHandlerFromEndpoint is same as RegisterAuthHandler but
 // automatically dials to "endpoint" and closes the connection when "ctx" gets done.
 func RegisterAuthHandlerFromEndpoint(ctx context.Context, mux *runtime.ServeMux, endpoint string, opts []grpc.DialOption) (err error) {

proto/prysm/v1alpha1/validator-client/web_api.proto

@@ -147,41 +147,6 @@ service SlashingProtection {
     }
 }
 
-// Health endpoints and log streaming will no longer be available, please use grafana and local log setups for reivew.
-// DEPRECATED: Prysm Web UI and associated endpoints will be fully removed in a future hard fork.
-service Health {
-    rpc GetBeaconNodeConnection(google.protobuf.Empty) returns (NodeConnectionResponse) {
-        option deprecated = true;
-        option (google.api.http) = {
-            get: "/v2/validator/health/node_connection"
-        };
-    }
-    rpc GetLogsEndpoints(google.protobuf.Empty) returns (LogsEndpointResponse) {
-        option deprecated = true;
-        option (google.api.http) = {
-            get: "/v2/validator/health/logs/endpoints"
-        };
-    }
-    rpc GetVersion(google.protobuf.Empty) returns (VersionResponse) {
-        option deprecated = true;
-        option (google.api.http) = {
-            get: "/v2/validator/health/version"
-        };
-    }
-    rpc StreamBeaconLogs(google.protobuf.Empty) returns (stream ethereum.eth.v1alpha1.LogsResponse) {
-        option deprecated = true;
-        option (google.api.http) = {
-            get: "/v2/validator/health/logs/beacon/stream"
-        };
-    }
-    rpc StreamValidatorLogs(google.protobuf.Empty) returns (stream ethereum.eth.v1alpha1.LogsResponse) {
-        option deprecated = true;
-        option (google.api.http) = {
-            get: "/v2/validator/health/logs/validator/stream"
-        };
-    }
-}
-
 // Web APIs such as the Keymanager APIs will no longer validate JWTs on the endpoint. Users should no longer expose the validator APIs to the public.
 // option deprecated = true; can't be added yet as it's used for keymanager API
 // DEPRECATED: Prysm Web UI and associated endpoints will be fully removed in a future hard fork.
@@ -330,36 +295,6 @@ message AccountRequest {
     repeated uint64 indices = 2;
 }
 
-// DEPRECATED: Prysm Web UI and associated endpoints will be fully removed in a future hard fork.
-message NodeConnectionResponse {
-    option deprecated = true;
-    // The host address of the beacon node the validator
-    // client is connected to.
-    string beacon_node_endpoint = 1;
-    // Whether the connection is active.
-    bool connected = 2;
-    // Whether the beacon node is currently synchronizing to chain head.
-    bool syncing = 3;
-    // The chain genesis time.
-    uint64 genesis_time = 4;
-    // Address of the validator deposit contract in the eth1 chain.
-    bytes deposit_contract_address = 5;
-}
-
-// DEPRECATED: Prysm Web UI and associated endpoints will be fully removed in a future hard fork.
-message LogsEndpointResponse {
-    option deprecated = true;
-	string validator_logs_endpoint = 1;
-	string beacon_logs_endpoint = 2;
-}
-
-// DEPRECATED: Prysm Web UI and associated endpoints will be fully removed in a future hard fork.
-message VersionResponse {
-    option deprecated = true;
-	string beacon = 1;
-	string validator = 2;
-}
-
 // DEPRECATED: Prysm Web UI and associated endpoints will be fully removed in a future hard fork.
 message HasWalletResponse {
     option deprecated = true;

validator/node/node.go

@@ -796,7 +796,6 @@ func (c *ValidatorClient) registerRPCGatewayService(router *mux.Router) error {
 		validatorpb.RegisterAuthHandler,
 		validatorpb.RegisterWalletHandler,
 		pb.RegisterHealthHandler,
-		validatorpb.RegisterHealthHandler,
 		validatorpb.RegisterAccountsHandler,
 		validatorpb.RegisterBeaconHandler,
 		validatorpb.RegisterSlashingProtectionHandler,

validator/rpc/BUILD.bazel

@@ -6,8 +6,8 @@ go_library(
         "accounts.go",
         "auth_token.go",
         "beacon.go",
+        "handlers_health.go",
         "handlers_keymanager.go",
-        "health.go",
         "intercepter.go",
         "log.go",
         "server.go",
@@ -91,8 +91,8 @@ go_test(
         "accounts_test.go",
         "auth_token_test.go",
         "beacon_test.go",
+        "handlers_health_test.go",
         "handlers_keymanager_test.go",
-        "health_test.go",
         "intercepter_test.go",
         "server_test.go",
         "slashing_test.go",
@@ -112,6 +112,7 @@ go_test(
         "//crypto/rand:go_default_library",
         "//encoding/bytesutil:go_default_library",
         "//io/file:go_default_library",
+        "//io/logs/mock:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
         "//proto/prysm/v1alpha1/validator-client:go_default_library",
         "//testing/assert:go_default_library",

validator/rpc/handlers_health.go

@@ -0,0 +1,163 @@
+package rpc
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+
+	http2 "github.com/prysmaticlabs/prysm/v4/network/http"
+	pb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/v4/runtime/version"
+	"go.opencensus.io/trace"
+	"google.golang.org/protobuf/types/known/emptypb"
+)
+
+// GetVersion returns the beacon node and validator client versions
+func (s *Server) GetVersion(w http.ResponseWriter, r *http.Request) {
+	ctx, span := trace.StartSpan(r.Context(), "validator.web.health.GetVersion")
+	defer span.End()
+
+	beacon, err := s.beaconNodeClient.GetVersion(ctx, &emptypb.Empty{})
+	if err != nil {
+		http2.HandleError(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	http2.WriteJson(w, struct {
+		Beacon    string `json:"beacon"`
+		Validator string `json:"validator"`
+	}{
+		Beacon:    beacon.Version,
+		Validator: version.Version(),
+	})
+}
+
+// StreamBeaconLogs from the beacon node via server-side events.
+func (s *Server) StreamBeaconLogs(w http.ResponseWriter, r *http.Request) {
+	// Wrap service context with a cancel in order to propagate the exiting of
+	// this method properly to the beacon node server.
+	ctx, span := trace.StartSpan(r.Context(), "validator.web.health.StreamBeaconLogs")
+	defer span.End()
+	// Set up SSE response headers
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+
+	// Flush helper function to ensure data is sent to client
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		http2.HandleError(w, "Streaming unsupported!", http.StatusInternalServerError)
+		return
+	}
+	// TODO: StreamBeaconLogs grpc will need to be replaced in the future
+	client, err := s.beaconNodeHealthClient.StreamBeaconLogs(ctx, &emptypb.Empty{})
+	if err != nil {
+		http2.HandleError(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	for {
+		select {
+		case <-s.ctx.Done():
+			return
+		case <-ctx.Done():
+			return
+		case <-client.Context().Done():
+			return
+		default:
+			logResp, err := client.Recv()
+			if err != nil {
+				http2.HandleError(w, "could not receive beacon logs from stream: "+err.Error(), http.StatusInternalServerError)
+				return
+			}
+			jsonResp, err := json.Marshal(logResp)
+			if err != nil {
+				http2.HandleError(w, "could not encode log response into JSON: "+err.Error(), http.StatusInternalServerError)
+				return
+			}
+
+			// Send the response as an SSE event
+			// Assuming resp has a String() method for simplicity
+			_, err = fmt.Fprintf(w, "%s\n", jsonResp)
+			if err != nil {
+				http2.HandleError(w, err.Error(), http.StatusInternalServerError)
+				return
+			}
+			// Flush the data to the client immediately
+			flusher.Flush()
+		}
+	}
+}
+
+// StreamValidatorLogs from the validator client via server-side events.
+func (s *Server) StreamValidatorLogs(w http.ResponseWriter, r *http.Request) {
+	ctx, span := trace.StartSpan(r.Context(), "validator.web.health.StreamValidatorLogs")
+	defer span.End()
+
+	// Ensure that the writer supports flushing.
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		http2.HandleError(w, "Streaming unsupported!", http.StatusInternalServerError)
+		return
+	}
+
+	ch := make(chan []byte, s.streamLogsBufferSize)
+	sub := s.logsStreamer.LogsFeed().Subscribe(ch)
+	defer func() {
+		sub.Unsubscribe()
+		close(ch)
+	}()
+	// Set up SSE response headers
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+
+	recentLogs := s.logsStreamer.GetLastFewLogs()
+	logStrings := make([]string, len(recentLogs))
+	for i, l := range recentLogs {
+		logStrings[i] = string(l)
+	}
+	ls := &pb.LogsResponse{
+		Logs: logStrings,
+	}
+	jsonLogs, err := json.Marshal(ls)
+	if err != nil {
+		http2.HandleError(w, "Failed to marshal logs: "+err.Error(), http.StatusInternalServerError)
+		return
+	}
+	_, err = fmt.Fprintf(w, "%s\n", jsonLogs)
+	if err != nil {
+		http2.HandleError(w, "Error sending data: "+err.Error(), http.StatusInternalServerError)
+		return
+	}
+	flusher.Flush()
+
+	for {
+		select {
+		case log := <-ch:
+			// Set up SSE response headers
+			ls = &pb.LogsResponse{
+				Logs: []string{string(log)},
+			}
+			jsonLogs, err = json.Marshal(ls)
+			if err != nil {
+				http2.HandleError(w, "Failed to marshal logs: "+err.Error(), http.StatusInternalServerError)
+				return
+			}
+			_, err = fmt.Fprintf(w, "%s\n", jsonLogs)
+			if err != nil {
+				http2.HandleError(w, "Error sending data: "+err.Error(), http.StatusInternalServerError)
+				return
+			}
+
+			flusher.Flush()
+		case <-s.ctx.Done():
+			return
+		case err := <-sub.Err():
+			http2.HandleError(w, "Subscriber error: "+err.Error(), http.StatusInternalServerError)
+			return
+		case <-ctx.Done():
+			return
+		}
+	}
+}

validator/rpc/handlers_health_test.go

@@ -0,0 +1,191 @@
+package rpc
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	"github.com/golang/protobuf/ptypes/empty"
+	"github.com/prysmaticlabs/prysm/v4/io/logs/mock"
+	"github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
+	pb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/v4/testing/require"
+	validatormock "github.com/prysmaticlabs/prysm/v4/testing/validator-mock"
+	"google.golang.org/grpc"
+)
+
+type MockBeaconNodeHealthClient struct {
+	grpc.ClientStream
+	logs []*pb.LogsResponse
+	err  error
+}
+
+func (m *MockBeaconNodeHealthClient) StreamBeaconLogs(ctx context.Context, in *empty.Empty, opts ...grpc.CallOption) (eth.Health_StreamBeaconLogsClient, error) {
+	return m, m.err
+}
+
+func (m *MockBeaconNodeHealthClient) Recv() (*eth.LogsResponse, error) {
+	if len(m.logs) == 0 {
+		return nil, io.EOF
+	}
+	log := m.logs[0]
+	m.logs = m.logs[1:]
+	return log, nil
+}
+
+func (m *MockBeaconNodeHealthClient) SendMsg(_ interface{}) error {
+	return m.err
+}
+
+func (m *MockBeaconNodeHealthClient) Context() context.Context {
+	return context.Background()
+}
+
+type flushableResponseRecorder struct {
+	*httptest.ResponseRecorder
+	flushed bool
+}
+
+func (f *flushableResponseRecorder) Flush() {
+	f.flushed = true
+}
+
+func TestStreamBeaconLogs(t *testing.T) {
+	logs := []*pb.LogsResponse{
+		{
+			Logs: []string{"log1", "log2"},
+		},
+		{
+			Logs: []string{"log3", "log4"},
+		},
+	}
+
+	mockClient := &MockBeaconNodeHealthClient{
+		logs: logs,
+		err:  nil,
+	}
+
+	// Setting up the mock in the server struct
+	s := Server{
+		ctx:                    context.Background(),
+		beaconNodeHealthClient: mockClient,
+	}
+
+	// Create a mock ResponseWriter and Request
+	w := &flushableResponseRecorder{
+		ResponseRecorder: httptest.NewRecorder(),
+	}
+	r := httptest.NewRequest("GET", "/v2/validator/health/logs/beacon/stream", nil)
+
+	// Call the function
+	s.StreamBeaconLogs(w, r)
+
+	// Assert the results
+	resp := w.Result()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("Expected status OK but got %v", resp.StatusCode)
+	}
+	ct, ok := resp.Header["Content-Type"]
+	require.Equal(t, ok, true)
+	require.Equal(t, ct[0], "text/event-stream")
+	cn, ok := resp.Header["Connection"]
+	require.Equal(t, ok, true)
+	require.Equal(t, cn[0], "keep-alive")
+	body, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	require.NotNil(t, body)
+	require.StringContains(t, `{"logs":["log1","log2"]}`, string(body))
+	require.StringContains(t, `{"logs":["log3","log4"]}`, string(body))
+	if !w.flushed {
+		t.Fatal("Flush was not called")
+	}
+}
+
+func TestStreamValidatorLogs(t *testing.T) {
+	ctx := context.Background()
+	mockLogs := [][]byte{
+		[]byte("[2023-10-31 10:00:00] INFO: Starting server..."),
+		[]byte("[2023-10-31 10:01:23] DEBUG: Database connection established."),
+		[]byte("[2023-10-31 10:05:45] WARN: High memory usage detected."),
+		[]byte("[2023-10-31 10:10:12] INFO: New user registered: user123."),
+		[]byte("[2023-10-31 10:15:30] ERROR: Failed to send email."),
+	}
+	logStreamer := mock.NewMockStreamer(mockLogs)
+	// Setting up the mock in the server struct
+	s := Server{
+		ctx:                  ctx,
+		logsStreamer:         logStreamer,
+		streamLogsBufferSize: 100,
+	}
+
+	w := &flushableResponseRecorder{
+		ResponseRecorder: httptest.NewRecorder(),
+	}
+	r := httptest.NewRequest("GET", "/v2/validator/health/logs/validator/stream", nil)
+	go func() {
+		s.StreamValidatorLogs(w, r)
+	}()
+	// wait for initiation of StreamValidatorLogs
+	time.Sleep(100 * time.Millisecond)
+	logStreamer.LogsFeed().Send([]byte("Some mock event data"))
+	// wait for feed
+	time.Sleep(100 * time.Millisecond)
+	s.ctx.Done()
+	// Assert the results
+	resp := w.Result()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("Expected status OK but got %v", resp.StatusCode)
+	}
+	ct, ok := resp.Header["Content-Type"]
+	require.Equal(t, ok, true)
+	require.Equal(t, ct[0], "text/event-stream")
+	cn, ok := resp.Header["Connection"]
+	require.Equal(t, ok, true)
+	require.Equal(t, cn[0], "keep-alive")
+	// Check if data was written
+	body, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	require.NotNil(t, body)
+
+	require.StringContains(t, `{"logs":["[2023-10-31 10:00:00] INFO: Starting server...","[2023-10-31 10:01:23] DEBUG: Database connection established.",`+
+		`"[2023-10-31 10:05:45] WARN: High memory usage detected.","[2023-10-31 10:10:12] INFO: New user registered: user123.","[2023-10-31 10:15:30] ERROR: Failed to send email."]}`, string(body))
+	require.StringContains(t, `{"logs":["Some mock event data"]}`, string(body))
+
+	// Check if Flush was called
+	if !w.flushed {
+		t.Fatal("Flush was not called")
+	}
+
+}
+
+func TestServer_GetVersion(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	ctx := context.Background()
+	mockNodeClient := validatormock.NewMockNodeClient(ctrl)
+	s := Server{
+		ctx:              ctx,
+		beaconNodeClient: mockNodeClient,
+	}
+	mockNodeClient.EXPECT().GetVersion(gomock.Any(), gomock.Any()).Return(&eth.Version{
+		Version:  "4.10.1",
+		Metadata: "beacon node",
+	}, nil)
+	r := httptest.NewRequest("GET", "/v2/validator/health/version", nil)
+	w := httptest.NewRecorder()
+	w.Body = &bytes.Buffer{}
+	s.GetVersion(w, r)
+	resp := w.Result()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("Expected status OK but got %v", resp.StatusCode)
+	}
+	body, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	require.NotNil(t, body)
+	require.StringContains(t, `{"beacon":"4.10.1","validator":"Prysm/Unknown/Local build. Built at: Moments ago"}`, string(body))
+}

validator/rpc/health.go

@@ -1,132 +0,0 @@
-package rpc
-
-import (
-	"context"
-	"time"
-
-	"github.com/pkg/errors"
-	pb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
-	validatorpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1/validator-client"
-	"github.com/prysmaticlabs/prysm/v4/runtime/version"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
-	"google.golang.org/protobuf/types/known/emptypb"
-)
-
-// GetBeaconNodeConnection retrieves the current beacon node connection
-// information, as well as its sync status.
-// DEPRECATED: Prysm Web UI and associated endpoints will be fully removed in a future hard fork.
-func (s *Server) GetBeaconNodeConnection(ctx context.Context, _ *emptypb.Empty) (*validatorpb.NodeConnectionResponse, error) {
-	syncStatus, err := s.syncChecker.Syncing(ctx)
-	if err != nil || s.validatorService.Status() != nil {
-		//nolint:nilerr
-		return &validatorpb.NodeConnectionResponse{
-			GenesisTime:        0,
-			BeaconNodeEndpoint: s.nodeGatewayEndpoint,
-			Connected:          false,
-			Syncing:            false,
-		}, nil
-	}
-	genesis, err := s.genesisFetcher.GenesisInfo(ctx)
-	if err != nil {
-		return nil, err
-	}
-	return &validatorpb.NodeConnectionResponse{
-		GenesisTime:            uint64(time.Unix(genesis.GenesisTime.Seconds, 0).Unix()),
-		DepositContractAddress: genesis.DepositContractAddress,
-		BeaconNodeEndpoint:     s.nodeGatewayEndpoint,
-		Connected:              true,
-		Syncing:                syncStatus,
-	}, nil
-}
-
-// GetLogsEndpoints for the beacon and validator client.
-// DEPRECATED: Prysm Web UI and associated endpoints will be fully removed in a future hard fork.
-func (*Server) GetLogsEndpoints(_ context.Context, _ *emptypb.Empty) (*validatorpb.LogsEndpointResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "unimplemented")
-}
-
-// GetVersion --
-// DEPRECATED: Prysm Web UI and associated endpoints will be fully removed in a future hard fork.
-func (s *Server) GetVersion(ctx context.Context, _ *emptypb.Empty) (*validatorpb.VersionResponse, error) {
-	beacon, err := s.beaconNodeClient.GetVersion(ctx, &emptypb.Empty{})
-	if err != nil {
-		return nil, err
-	}
-
-	return &validatorpb.VersionResponse{
-		Beacon:    beacon.Version,
-		Validator: version.Version(),
-	}, nil
-}
-
-// StreamBeaconLogs from the beacon node via a gRPC server-side stream.
-// DEPRECATED: Prysm Web UI and associated endpoints will be fully removed in a future hard fork.
-func (s *Server) StreamBeaconLogs(req *emptypb.Empty, stream validatorpb.Health_StreamBeaconLogsServer) error {
-	// Wrap service context with a cancel in order to propagate the exiting of
-	// this method properly to the beacon node server.
-	ctx, cancel := context.WithCancel(s.ctx)
-	defer cancel()
-
-	client, err := s.beaconNodeHealthClient.StreamBeaconLogs(ctx, req)
-	if err != nil {
-		return err
-	}
-	for {
-		select {
-		case <-s.ctx.Done():
-			return status.Error(codes.Canceled, "Context canceled")
-		case <-stream.Context().Done():
-			return status.Error(codes.Canceled, "Context canceled")
-		case <-client.Context().Done():
-			return status.Error(codes.Canceled, "Context canceled")
-		default:
-			resp, err := client.Recv()
-			if err != nil {
-				return errors.Wrap(err, "could not receive beacon logs from stream")
-			}
-			if err := stream.Send(resp); err != nil {
-				return status.Errorf(codes.Unavailable, "Could not send over stream: %v", err)
-			}
-		}
-	}
-}
-
-// StreamValidatorLogs from the validator client via a gRPC server-side stream.
-// DEPRECATED: Prysm Web UI and associated endpoints will be fully removed in a future hard fork.
-func (s *Server) StreamValidatorLogs(_ *emptypb.Empty, stream validatorpb.Health_StreamValidatorLogsServer) error {
-	ch := make(chan []byte, s.streamLogsBufferSize)
-	sub := s.logsStreamer.LogsFeed().Subscribe(ch)
-	defer func() {
-		sub.Unsubscribe()
-		defer close(ch)
-	}()
-
-	recentLogs := s.logsStreamer.GetLastFewLogs()
-	logStrings := make([]string, len(recentLogs))
-	for i, log := range recentLogs {
-		logStrings[i] = string(log)
-	}
-	if err := stream.Send(&pb.LogsResponse{
-		Logs: logStrings,
-	}); err != nil {
-		return status.Errorf(codes.Unavailable, "Could not send over stream: %v", err)
-	}
-	for {
-		select {
-		case log := <-ch:
-			resp := &pb.LogsResponse{
-				Logs: []string{string(log)},
-			}
-			if err := stream.Send(resp); err != nil {
-				return status.Errorf(codes.Unavailable, "Could not send over stream: %v", err)
-			}
-		case <-s.ctx.Done():
-			return status.Error(codes.Canceled, "Context canceled")
-		case err := <-sub.Err():
-			return status.Errorf(codes.Canceled, "Subscriber error, closing: %v", err)
-		case <-stream.Context().Done():
-			return status.Error(codes.Canceled, "Context canceled")
-		}
-	}
-}

validator/rpc/health_test.go

@@ -1,54 +0,0 @@
-package rpc
-
-import (
-	"context"
-	"testing"
-	"time"
-
-	"github.com/golang/protobuf/ptypes/empty"
-	ethpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
-	pb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1/validator-client"
-	"github.com/prysmaticlabs/prysm/v4/testing/require"
-	"github.com/prysmaticlabs/prysm/v4/validator/client"
-	"google.golang.org/protobuf/types/known/timestamppb"
-)
-
-type mockSyncChecker struct {
-	syncing bool
-}
-
-func (m *mockSyncChecker) Syncing(_ context.Context) (bool, error) {
-	return m.syncing, nil
-}
-
-type mockGenesisFetcher struct{}
-
-func (_ *mockGenesisFetcher) GenesisInfo(_ context.Context) (*ethpb.Genesis, error) {
-	genesis := timestamppb.New(time.Unix(0, 0))
-	return &ethpb.Genesis{
-		GenesisTime: genesis,
-	}, nil
-}
-
-func TestServer_GetBeaconNodeConnection(t *testing.T) {
-	ctx := context.Background()
-	endpoint := "localhost:90210"
-	vs, err := client.NewValidatorService(ctx, &client.Config{})
-	require.NoError(t, err)
-	s := &Server{
-		walletInitialized:   true,
-		validatorService:    vs,
-		syncChecker:         &mockSyncChecker{syncing: false},
-		genesisFetcher:      &mockGenesisFetcher{},
-		nodeGatewayEndpoint: endpoint,
-	}
-	got, err := s.GetBeaconNodeConnection(ctx, &empty.Empty{})
-	require.NoError(t, err)
-	want := &pb.NodeConnectionResponse{
-		BeaconNodeEndpoint: endpoint,
-		Connected:          true,
-		Syncing:            false,
-		GenesisTime:        uint64(time.Unix(0, 0).Unix()),
-	}
-	require.DeepEqual(t, want, got)
-}

validator/rpc/server.go

@@ -185,7 +185,6 @@ func (s *Server) Start() {
 	reflection.Register(s.grpcServer)
 	validatorpb.RegisterAuthServer(s.grpcServer, s)
 	validatorpb.RegisterWalletServer(s.grpcServer, s)
-	validatorpb.RegisterHealthServer(s.grpcServer, s)
 	validatorpb.RegisterBeaconServer(s.grpcServer, s)
 	validatorpb.RegisterAccountsServer(s.grpcServer, s)
 	validatorpb.RegisterSlashingProtectionServer(s.grpcServer, s)
@@ -234,7 +233,10 @@ func (s *Server) InitializeRoutes() error {
 	s.router.HandleFunc("/eth/v1/validator/{pubkey}/feerecipient", s.SetFeeRecipientByPubkey).Methods(http.MethodPost)
 	s.router.HandleFunc("/eth/v1/validator/{pubkey}/feerecipient", s.DeleteFeeRecipientByPubkey).Methods(http.MethodDelete)
 	s.router.HandleFunc("/eth/v1/validator/{pubkey}/voluntary_exit", s.SetVoluntaryExit).Methods(http.MethodPost)
-	// ...
+	// web health endpoints
+	s.router.HandleFunc("/v2/validator/health/version", s.GetVersion).Methods(http.MethodGet)
+	s.router.HandleFunc("/v2/validator/health/logs/validator/stream", s.StreamValidatorLogs).Methods(http.MethodGet)
+	s.router.HandleFunc("/v2/validator/health/logs/beacon/stream", s.StreamBeaconLogs).Methods(http.MethodGet)
 	log.Info("Initialized REST API routes")
 	return nil
 }

validator/rpc/server_test.go

@@ -19,11 +19,14 @@ func TestServer_InitializeRoutes(t *testing.T) {
 	require.NoError(t, err)
 
 	wantRouteList := map[string][]string{
-		"/eth/v1/keystores":                         {http.MethodGet, http.MethodPost, http.MethodDelete},
+		"/eth/v1/keystores":                          {http.MethodGet, http.MethodPost, http.MethodDelete},
-		"/eth/v1/remotekeys":                        {http.MethodGet, http.MethodPost, http.MethodDelete},
+		"/eth/v1/remotekeys":                         {http.MethodGet, http.MethodPost, http.MethodDelete},
-		"/eth/v1/validator/{pubkey}/gas_limit":      {http.MethodGet, http.MethodPost, http.MethodDelete},
+		"/eth/v1/validator/{pubkey}/gas_limit":       {http.MethodGet, http.MethodPost, http.MethodDelete},
-		"/eth/v1/validator/{pubkey}/feerecipient":   {http.MethodGet, http.MethodPost, http.MethodDelete},
+		"/eth/v1/validator/{pubkey}/feerecipient":    {http.MethodGet, http.MethodPost, http.MethodDelete},
-		"/eth/v1/validator/{pubkey}/voluntary_exit": {http.MethodPost},
+		"/eth/v1/validator/{pubkey}/voluntary_exit":  {http.MethodPost},
+		"/v2/validator/health/version":               {http.MethodGet},
+		"/v2/validator/health/logs/validator/stream": {http.MethodGet},
+		"/v2/validator/health/logs/beacon/stream":    {http.MethodGet},
 	}
 	gotRouteList := make(map[string][]string)
 	err = s.router.Walk(func(route *mux.Route, router *mux.Router, ancestors []*mux.Route) error {