-
Notifications
You must be signed in to change notification settings - Fork 2.4k
/
grpc.go
104 lines (84 loc) · 3.18 KB
/
grpc.go
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
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
// Copyright (c) 2018 The Jaeger Authors.
// SPDX-License-Identifier: Apache-2.0
package server
import (
"fmt"
"net"
"time"
"go.uber.org/zap"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/health"
"google.golang.org/grpc/health/grpc_health_v1"
"google.golang.org/grpc/keepalive"
"google.golang.org/grpc/reflection"
"github.com/jaegertracing/jaeger/cmd/collector/app/handler"
"github.com/jaegertracing/jaeger/cmd/collector/app/sampling"
"github.com/jaegertracing/jaeger/cmd/collector/app/sampling/samplingstrategy"
"github.com/jaegertracing/jaeger/pkg/config/tlscfg"
"github.com/jaegertracing/jaeger/proto-gen/api_v2"
)
// GRPCServerParams to construct a new Jaeger Collector gRPC Server
type GRPCServerParams struct {
TLSConfig tlscfg.Options
HostPort string
Handler *handler.GRPCHandler
SamplingProvider samplingstrategy.Provider
Logger *zap.Logger
OnError func(error)
MaxReceiveMessageLength int
MaxConnectionAge time.Duration
MaxConnectionAgeGrace time.Duration
// Set by the server to indicate the actual host:port of the server.
HostPortActual string
}
// StartGRPCServer based on the given parameters
func StartGRPCServer(params *GRPCServerParams) (*grpc.Server, error) {
var server *grpc.Server
var grpcOpts []grpc.ServerOption
if params.MaxReceiveMessageLength > 0 {
grpcOpts = append(grpcOpts, grpc.MaxRecvMsgSize(params.MaxReceiveMessageLength))
}
grpcOpts = append(grpcOpts, grpc.KeepaliveParams(keepalive.ServerParameters{
MaxConnectionAge: params.MaxConnectionAge,
MaxConnectionAgeGrace: params.MaxConnectionAgeGrace,
}))
if params.TLSConfig.Enabled {
// user requested a server with TLS, setup creds
tlsCfg, err := params.TLSConfig.Config(params.Logger)
if err != nil {
return nil, err
}
creds := credentials.NewTLS(tlsCfg)
grpcOpts = append(grpcOpts, grpc.Creds(creds))
}
server = grpc.NewServer(grpcOpts...)
reflection.Register(server)
listener, err := net.Listen("tcp", params.HostPort)
if err != nil {
return nil, fmt.Errorf("failed to listen on gRPC port: %w", err)
}
params.HostPortActual = listener.Addr().String()
if err := serveGRPC(server, listener, params); err != nil {
return nil, err
}
return server, nil
}
func serveGRPC(server *grpc.Server, listener net.Listener, params *GRPCServerParams) error {
healthServer := health.NewServer()
api_v2.RegisterCollectorServiceServer(server, params.Handler)
api_v2.RegisterSamplingManagerServer(server, sampling.NewGRPCHandler(params.SamplingProvider))
healthServer.SetServingStatus("jaeger.api_v2.CollectorService", grpc_health_v1.HealthCheckResponse_SERVING)
healthServer.SetServingStatus("jaeger.api_v2.SamplingManager", grpc_health_v1.HealthCheckResponse_SERVING)
grpc_health_v1.RegisterHealthServer(server, healthServer)
params.Logger.Info("Starting jaeger-collector gRPC server", zap.String("grpc.host-port", params.HostPortActual))
go func() {
if err := server.Serve(listener); err != nil {
params.Logger.Error("Could not launch gRPC service", zap.Error(err))
if params.OnError != nil {
params.OnError(err)
}
}
}()
return nil
}