-
Notifications
You must be signed in to change notification settings - Fork 55
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Refactors otlpserver to an interface and adds a simple HTTP server that can receive and translate traces. This is step 1 to do #142 correctly, as I wish to test the heck out of endpoint and protocol combinations in the next couple PRs. * rename NewServer to NewGrpcServer * rename gRPC server functions & variables * rename server.go -> grpcserver.go * fix double import of trace proto * refactor grpc server into an interface and implementation Broke the surface otel-cli and main_test care about into an interface. Ported grpcserver and referents to the interface. Added skeleton http server. Tests pass :) * refactor code further, adding protocol to fixtures Added a simple http handler for testing. Not ready yet. Added plumbing so http protocol can be requested, but it's not used yet. Correct some types. * create a failing http test * get http traces working Broke the code that converts protobuf data to CliEvents out into its own function and moved it to clievent.go. Converted the gRPC server to use that. Added it to the HTTP server and it just worked the first time. Since proto is now imported directly, it go.mod changed to reflect that. * add a way to test http path, proto, etc. Added ServerMeta to the CliEvent struct as a string map that httpserver now plugs a bunch of metadata into. Added that to the test harness. Added metadata to the test so now path & proto are verified. * add proto field go grpc server_meta * fix test * fix comment * add godoc to otelToCliEvent()
- Loading branch information
Showing
9 changed files
with
313 additions
and
125 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
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,92 @@ | ||
| package otlpserver | ||
|
|
||
| import ( | ||
| "context" | ||
| "log" | ||
| "net" | ||
| "strings" | ||
| "sync" | ||
|
|
||
| v1 "go.opentelemetry.io/proto/otlp/collector/trace/v1" | ||
|
|
||
| "google.golang.org/grpc" | ||
| ) | ||
|
|
||
| // GrpcServer is a gRPC/OTLP server handle. | ||
| type GrpcServer struct { | ||
| server *grpc.Server | ||
| callback Callback | ||
| stoponce sync.Once | ||
| stopper chan struct{} | ||
| stopdone chan struct{} | ||
| doneonce sync.Once | ||
| v1.UnimplementedTraceServiceServer | ||
| } | ||
|
|
||
| // NewGrpcServer takes a callback and stop function and returns a Server ready | ||
| // to run with .Serve(). | ||
| func NewGrpcServer(cb Callback, stop Stopper) *GrpcServer { | ||
| s := GrpcServer{ | ||
| server: grpc.NewServer(), | ||
| callback: cb, | ||
| stopper: make(chan struct{}), | ||
| stopdone: make(chan struct{}, 1), | ||
| } | ||
|
|
||
| v1.RegisterTraceServiceServer(s.server, &s) | ||
|
|
||
| // single place to stop the server, used by timeout and max-spans | ||
| go func() { | ||
| <-s.stopper | ||
| stop(&s) | ||
| s.server.GracefulStop() | ||
| }() | ||
|
|
||
| return &s | ||
| } | ||
|
|
||
| // ServeGRPC takes a listener and starts the GRPC server on that listener. | ||
| // Blocks until Stop() is called. | ||
| func (gs *GrpcServer) Serve(listener net.Listener) error { | ||
| err := gs.server.Serve(listener) | ||
| gs.stopdone <- struct{}{} | ||
| return err | ||
| } | ||
|
|
||
| // ListenAndServeGRPC starts a TCP listener then starts the GRPC server using | ||
| // ServeGRPC for you. | ||
| func (gs *GrpcServer) ListenAndServe(otlpEndpoint string) { | ||
| otlpEndpoint = strings.TrimPrefix(otlpEndpoint, "grpc://") | ||
| listener, err := net.Listen("tcp", otlpEndpoint) | ||
| if err != nil { | ||
| log.Fatalf("failed to listen on OTLP endpoint %q: %s", otlpEndpoint, err) | ||
| } | ||
| if err := gs.Serve(listener); err != nil { | ||
| log.Fatalf("failed to serve: %s", err) | ||
| } | ||
| } | ||
|
|
||
| // Stop sends a value to the server shutdown goroutine so it stops GRPC | ||
| // and calls the stop function given to newServer. Safe to call multiple times. | ||
| func (gs *GrpcServer) Stop() { | ||
| gs.stoponce.Do(func() { | ||
| gs.stopper <- struct{}{} | ||
| }) | ||
| } | ||
|
|
||
| // StopWait stops the server and waits for it to affirm shutdown. | ||
| func (gs *GrpcServer) StopWait() { | ||
| gs.Stop() | ||
| gs.doneonce.Do(func() { | ||
| <-gs.stopdone | ||
| }) | ||
| } | ||
|
|
||
| // Export implements the gRPC server interface for exporting messages. | ||
| func (gs *GrpcServer) Export(ctx context.Context, req *v1.ExportTraceServiceRequest) (*v1.ExportTraceServiceResponse, error) { | ||
| done := otelToCliEvent(gs.callback, req, map[string]string{"proto": "grpc"}) | ||
| if done { | ||
| go gs.StopWait() | ||
| } | ||
| return &v1.ExportTraceServiceResponse{}, nil | ||
| } |
Oops, something went wrong.