-
Notifications
You must be signed in to change notification settings - Fork 2.4k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add gRPC communication between agent and collector (#1165)
* Add gRPC communication between agent and collector Signed-off-by: Pavol Loffay <ploffay@redhat.com> * uncomment make and fix comment Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Fix grpc handler initialization Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Copy process if missing Signed-off-by: Pavol Loffay <ploffay@redhat.com> * log only warn and error from grpc Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Fix review comments Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Add todo Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Add Roundrobin grpc load balancer Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Fix comment Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Fix review comments Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Change grpc target name Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Defaul return error Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Fix back tchannel collector addr Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Remove collector from grpc reporter flag Signed-off-by: Pavol Loffay <ploffay@redhat.com>
- Loading branch information
1 parent
f2eb7d1
commit 9635a33
Showing
43 changed files
with
3,967 additions
and
1,259 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| // Copyright (c) 2018 The Jaeger Authors. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package reporter | ||
|
|
||
| import ( | ||
| "flag" | ||
| "fmt" | ||
|
|
||
| "github.com/spf13/viper" | ||
| ) | ||
|
|
||
| const ( | ||
| reporterType = "reporter.type" | ||
| // TCHANNEL is name of tchannel reporter. | ||
| TCHANNEL Type = "tchannel" | ||
| // GRPC is name of gRPC reporter. | ||
| GRPC Type = "grpc" | ||
| ) | ||
|
|
||
| // Type defines type of reporter. | ||
| type Type string | ||
|
|
||
| // Options holds generic reporter configuration. | ||
| type Options struct { | ||
| ReporterType Type | ||
| } | ||
|
|
||
| // AddFlags adds flags for Options. | ||
| func AddFlags(flags *flag.FlagSet) { | ||
| flags.String(reporterType, string(TCHANNEL), fmt.Sprintf("Reporter type to use e.g. %s, %s", string(TCHANNEL), string(GRPC))) | ||
| } | ||
|
|
||
| // InitFromViper initializes Options with properties retrieved from Viper. | ||
| func (b *Options) InitFromViper(v *viper.Viper) *Options { | ||
| b.ReporterType = Type(v.GetString(reporterType)) | ||
| return b | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| // Copyright (c) 2018 The Jaeger Authors. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package reporter | ||
|
|
||
| import ( | ||
| "flag" | ||
| "testing" | ||
|
|
||
| "github.com/spf13/cobra" | ||
| "github.com/spf13/viper" | ||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestBingFlags(t *testing.T) { | ||
| v := viper.New() | ||
| command := cobra.Command{} | ||
| flags := &flag.FlagSet{} | ||
| AddFlags(flags) | ||
| command.PersistentFlags().AddGoFlagSet(flags) | ||
| v.BindPFlags(command.PersistentFlags()) | ||
|
|
||
| err := command.ParseFlags([]string{ | ||
| "--reporter.type=grpc", | ||
| }) | ||
| require.NoError(t, err) | ||
|
|
||
| b := &Options{} | ||
| b.InitFromViper(v) | ||
| assert.Equal(t, Type("grpc"), b.ReporterType) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| // Copyright (c) 2018 The Jaeger Authors. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package grpc | ||
|
|
||
| import ( | ||
| "go.uber.org/zap" | ||
| "google.golang.org/grpc" | ||
| "google.golang.org/grpc/balancer/roundrobin" | ||
| "google.golang.org/grpc/resolver" | ||
| "google.golang.org/grpc/resolver/manual" | ||
|
|
||
| "github.com/jaegertracing/jaeger/cmd/agent/app/httpserver" | ||
| aReporter "github.com/jaegertracing/jaeger/cmd/agent/app/reporter" | ||
| ) | ||
|
|
||
| // ProxyBuilder holds objects communicating with collector | ||
| type ProxyBuilder struct { | ||
| reporter aReporter.Reporter | ||
| manager httpserver.ClientConfigManager | ||
| } | ||
|
|
||
| // NewCollectorProxy creates ProxyBuilder | ||
| func NewCollectorProxy(o *Options, logger *zap.Logger) *ProxyBuilder { | ||
| // It does not return error if the collector is not running | ||
| // a way to fail immediately is to call WithBlock and WithTimeout | ||
| var conn *grpc.ClientConn | ||
| if len(o.CollectorHostPort) > 1 { | ||
| r, _ := manual.GenerateAndRegisterManualResolver() | ||
| var resolvedAddrs []resolver.Address | ||
| for _, addr := range o.CollectorHostPort { | ||
| resolvedAddrs = append(resolvedAddrs, resolver.Address{Addr: addr}) | ||
| } | ||
| r.InitialAddrs(resolvedAddrs) | ||
| conn, _ = grpc.Dial(r.Scheme()+":///round_robin", grpc.WithInsecure(), grpc.WithBalancerName(roundrobin.Name)) | ||
| } else { | ||
| conn, _ = grpc.Dial(o.CollectorHostPort[0], grpc.WithInsecure()) | ||
| } | ||
| return &ProxyBuilder{ | ||
| reporter: NewReporter(conn, logger), | ||
| manager: NewSamplingManager(conn)} | ||
| } | ||
|
|
||
| // GetReporter returns Reporter | ||
| func (b ProxyBuilder) GetReporter() aReporter.Reporter { | ||
| return b.reporter | ||
| } | ||
|
|
||
| // GetManager returns manager | ||
| func (b ProxyBuilder) GetManager() httpserver.ClientConfigManager { | ||
| return b.manager | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| // Copyright (c) 2018 The Jaeger Authors. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package grpc | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
| "go.uber.org/zap" | ||
| "google.golang.org/grpc" | ||
|
|
||
| "github.com/jaegertracing/jaeger/proto-gen/api_v2" | ||
| "github.com/jaegertracing/jaeger/thrift-gen/jaeger" | ||
| ) | ||
|
|
||
| func TestProxyBuilder(t *testing.T) { | ||
| proxy := NewCollectorProxy(&Options{CollectorHostPort: []string{"localhost:0000"}}, zap.NewNop()) | ||
| require.NotNil(t, proxy) | ||
| assert.NotNil(t, proxy.GetReporter()) | ||
| assert.NotNil(t, proxy.GetManager()) | ||
| } | ||
|
|
||
| func TestMultipleCollectors(t *testing.T) { | ||
| spanHandler1 := &mockSpanHandler{} | ||
| s1, addr1 := initializeGRPCTestServer(t, func(s *grpc.Server) { | ||
| api_v2.RegisterCollectorServiceServer(s, spanHandler1) | ||
| }) | ||
| defer s1.Stop() | ||
| spanHandler2 := &mockSpanHandler{} | ||
| s2, addr2 := initializeGRPCTestServer(t, func(s *grpc.Server) { | ||
| api_v2.RegisterCollectorServiceServer(s, spanHandler2) | ||
| }) | ||
| defer s2.Stop() | ||
|
|
||
| proxy := NewCollectorProxy(&Options{CollectorHostPort: []string{addr1.String(), addr2.String()}}, zap.NewNop()) | ||
| require.NotNil(t, proxy) | ||
| assert.NotNil(t, proxy.GetReporter()) | ||
| assert.NotNil(t, proxy.GetManager()) | ||
|
|
||
| var bothServers = false | ||
| // TODO do not iterate, just create two batches | ||
| for i := 0; i < 10; i++ { | ||
| r := proxy.GetReporter() | ||
| err := r.EmitBatch(&jaeger.Batch{Spans: []*jaeger.Span{{OperationName: "op"}}, Process: &jaeger.Process{ServiceName: "service"}}) | ||
| require.NoError(t, err) | ||
| if len(spanHandler1.getRequests()) > 0 && len(spanHandler2.getRequests()) > 0 { | ||
| bothServers = true | ||
| break | ||
| } | ||
| } | ||
| assert.Equal(t, true, bothServers) | ||
| } |
Oops, something went wrong.