Skip to content

Commit

Permalink
Add /api/sampling endpoint in collector
Browse files Browse the repository at this point in the history
Return metricsFactory from test server init

fix import structure

fix unwanted change in agent server

Add license on new file

Signed-off-by: RickyRajinder <singh.sangh@gmail.com>

Add tests for errors

Refactor collector API handler to extend agent handler

Remove unused fields

Unexport fields in HttpHandler

Update callers of NewAPIHandler

Run go fmt

Fix lint issues

Fix spelling error in comment

Refactor HTTP Handler to shared location

Implement clientConfigManager interface for collector

Run gofmt

Resolve prior dependencies

Rename HTTPHandler

Update comments and fix failing test in agent

Change mux version

Add empty test

Rebase master and register routes in APIHandler

Remove empty test and rename handler variables

Run fmt
  • Loading branch information
RickyRajinder committed Jan 1, 2020
1 parent c6677ca commit 7350001
Show file tree
Hide file tree
Showing 5 changed files with 66 additions and 21 deletions.
23 changes: 22 additions & 1 deletion cmd/all-in-one/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
package main

import (
"errors"
"fmt"
"io"
"log"
Expand Down Expand Up @@ -53,6 +54,7 @@ import (
"github.com/jaegertracing/jaeger/cmd/flags"
queryApp "github.com/jaegertracing/jaeger/cmd/query/app"
"github.com/jaegertracing/jaeger/cmd/query/app/querysvc"
clientcfgHandler "github.com/jaegertracing/jaeger/pkg/clientcfg/clientcfghttp"
"github.com/jaegertracing/jaeger/pkg/config"
"github.com/jaegertracing/jaeger/pkg/healthcheck"
"github.com/jaegertracing/jaeger/pkg/recoveryhandler"
Expand All @@ -64,6 +66,7 @@ import (
"github.com/jaegertracing/jaeger/storage/dependencystore"
"github.com/jaegertracing/jaeger/storage/spanstore"
storageMetrics "github.com/jaegertracing/jaeger/storage/spanstore/metrics"
"github.com/jaegertracing/jaeger/thrift-gen/baggage"
jc "github.com/jaegertracing/jaeger/thrift-gen/jaeger"
sc "github.com/jaegertracing/jaeger/thrift-gen/sampling"
zc "github.com/jaegertracing/jaeger/thrift-gen/zipkincore"
Expand Down Expand Up @@ -251,7 +254,13 @@ func startCollector(

{
r := mux.NewRouter()
apiHandler := collectorApp.NewAPIHandler(jaegerBatchesHandler)
configManager := &collectorConfigManager{samplingStrategy: strategyStore}
cfgParams := clientcfgHandler.HTTPHandlerParams{
ConfigManager: configManager,
MetricsFactory: metricsFactory,
LegacySamplingEndpoint: false,
}
apiHandler := collectorApp.NewAPIHandler(jaegerBatchesHandler, *clientcfgHandler.NewHTTPHandler(cfgParams))
apiHandler.RegisterRoutes(r)
httpPortStr := ":" + strconv.Itoa(cOpts.CollectorHTTPPort)
recoveryHandler := recoveryhandler.NewRecoveryHandler(logger, true)
Expand Down Expand Up @@ -322,6 +331,18 @@ func startQuery(
return server
}

type collectorConfigManager struct {
samplingStrategy strategystore.StrategyStore
}

func (c *collectorConfigManager) GetSamplingStrategy(serviceName string) (*sc.SamplingStrategyResponse, error) {
return c.samplingStrategy.GetSamplingStrategy(serviceName)
}

func (c *collectorConfigManager) GetBaggageRestrictions(serviceName string) ([]*baggage.BaggageRestriction, error) {
return nil, errors.New("baggage not implemented")
}

func initSamplingStrategyStore(
samplingStrategyStoreFactory *ss.Factory,
metricsFactory metrics.Factory,
Expand Down
22 changes: 12 additions & 10 deletions cmd/collector/app/http_handler.go
Original file line number Diff line number Diff line change
Expand Up @@ -24,14 +24,10 @@ import (
"github.com/apache/thrift/lib/go/thrift"
"github.com/gorilla/mux"

"github.com/jaegertracing/jaeger/pkg/clientcfg/clientcfghttp"
tJaeger "github.com/jaegertracing/jaeger/thrift-gen/jaeger"
)

const (
// UnableToReadBodyErrFormat is an error message for invalid requests
UnableToReadBodyErrFormat = "Unable to process request body: %v"
)

var (
acceptedThriftFormats = map[string]struct{}{
"application/x-thrift": {},
Expand All @@ -42,19 +38,20 @@ var (
// APIHandler handles all HTTP calls to the collector
type APIHandler struct {
jaegerBatchesHandler JaegerBatchesHandler
cfgHandler clientcfghttp.HTTPHandler
}

// NewAPIHandler returns a new APIHandler
func NewAPIHandler(
jaegerBatchesHandler JaegerBatchesHandler,
) *APIHandler {
func NewAPIHandler(jaegerBatchesHandler JaegerBatchesHandler, cfgHandler clientcfghttp.HTTPHandler) *APIHandler {
return &APIHandler{
jaegerBatchesHandler: jaegerBatchesHandler,
cfgHandler: cfgHandler,
}
}

// RegisterRoutes registers routes for this handler on the given router
func (aH *APIHandler) RegisterRoutes(router *mux.Router) {
aH.cfgHandler.RegisterRoutes(router)
router.HandleFunc("/api/traces", aH.SaveSpan).Methods(http.MethodPost)
}

Expand All @@ -63,7 +60,7 @@ func (aH *APIHandler) SaveSpan(w http.ResponseWriter, r *http.Request) {
bodyBytes, err := ioutil.ReadAll(r.Body)
r.Body.Close()
if err != nil {
http.Error(w, fmt.Sprintf(UnableToReadBodyErrFormat, err), http.StatusInternalServerError)
http.Error(w, ErrUnableToReadBody(err), http.StatusInternalServerError)
return
}

Expand All @@ -83,7 +80,7 @@ func (aH *APIHandler) SaveSpan(w http.ResponseWriter, r *http.Request) {
// (NB): We decided to use this struct instead of straight batches to be as consistent with tchannel intake as possible.
batch := &tJaeger.Batch{}
if err = tdes.Read(batch, bodyBytes); err != nil {
http.Error(w, fmt.Sprintf(UnableToReadBodyErrFormat, err), http.StatusBadRequest)
http.Error(w, ErrUnableToReadBody(err), http.StatusBadRequest)
return
}
batches := []*tJaeger.Batch{batch}
Expand All @@ -95,3 +92,8 @@ func (aH *APIHandler) SaveSpan(w http.ResponseWriter, r *http.Request) {

w.WriteHeader(http.StatusAccepted)
}

// ErrUnableToReadBody returns a formatted error message
func ErrUnableToReadBody(err error) string {
return fmt.Sprintf("Unable to process request body: %v", err)
}
5 changes: 3 additions & 2 deletions cmd/collector/app/http_handler_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ import (
jaegerClient "github.com/uber/jaeger-client-go"
"github.com/uber/jaeger-client-go/transport"

"github.com/jaegertracing/jaeger/pkg/clientcfg/clientcfghttp"
"github.com/jaegertracing/jaeger/thrift-gen/jaeger"
)

Expand All @@ -57,7 +58,7 @@ func (p *mockJaegerHandler) getBatches() []*jaeger.Batch {

func initializeTestServer(err error) (*httptest.Server, *APIHandler) {
r := mux.NewRouter()
handler := NewAPIHandler(&mockJaegerHandler{err: err})
handler := NewAPIHandler(&mockJaegerHandler{err: err}, clientcfghttp.HTTPHandler{})
handler.RegisterRoutes(r)
return httptest.NewServer(r), handler
}
Expand Down Expand Up @@ -158,7 +159,7 @@ func TestMalformedFormat(t *testing.T) {
}

func TestCannotReadBodyFromRequest(t *testing.T) {
handler := NewAPIHandler(&mockJaegerHandler{})
handler := NewAPIHandler(&mockJaegerHandler{}, clientcfghttp.HTTPHandler{})
req, err := http.NewRequest(http.MethodPost, "whatever", &errReader{})
assert.NoError(t, err)
rw := dummyResponseWriter{}
Expand Down
12 changes: 6 additions & 6 deletions cmd/collector/app/zipkin/http_handler.go
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ func (aH *APIHandler) saveSpans(w http.ResponseWriter, r *http.Request) {
if strings.Contains(r.Header.Get("Content-Encoding"), "gzip") {
gz, err := gunzip(bRead)
if err != nil {
http.Error(w, fmt.Sprintf(app.UnableToReadBodyErrFormat, err), http.StatusBadRequest)
http.Error(w, app.ErrUnableToReadBody(err), http.StatusBadRequest)
return
}
defer gz.Close()
Expand All @@ -77,7 +77,7 @@ func (aH *APIHandler) saveSpans(w http.ResponseWriter, r *http.Request) {

bodyBytes, err := ioutil.ReadAll(bRead)
if err != nil {
http.Error(w, fmt.Sprintf(app.UnableToReadBodyErrFormat, err), http.StatusInternalServerError)
http.Error(w, app.ErrUnableToReadBody(err), http.StatusInternalServerError)
return
}

Expand All @@ -99,7 +99,7 @@ func (aH *APIHandler) saveSpans(w http.ResponseWriter, r *http.Request) {
return
}
if err != nil {
http.Error(w, fmt.Sprintf(app.UnableToReadBodyErrFormat, err), http.StatusBadRequest)
http.Error(w, app.ErrUnableToReadBody(err), http.StatusBadRequest)
return
}

Expand All @@ -117,7 +117,7 @@ func (aH *APIHandler) saveSpansV2(w http.ResponseWriter, r *http.Request) {
if strings.Contains(r.Header.Get("Content-Encoding"), "gzip") {
gz, err := gunzip(bRead)
if err != nil {
http.Error(w, fmt.Sprintf(app.UnableToReadBodyErrFormat, err), http.StatusBadRequest)
http.Error(w, app.ErrUnableToReadBody(err), http.StatusBadRequest)
return
}
defer gz.Close()
Expand All @@ -126,7 +126,7 @@ func (aH *APIHandler) saveSpansV2(w http.ResponseWriter, r *http.Request) {

bodyBytes, err := ioutil.ReadAll(bRead)
if err != nil {
http.Error(w, fmt.Sprintf(app.UnableToReadBodyErrFormat, err), http.StatusInternalServerError)
http.Error(w, app.ErrUnableToReadBody(err), http.StatusInternalServerError)
return
}

Expand All @@ -149,7 +149,7 @@ func (aH *APIHandler) saveSpansV2(w http.ResponseWriter, r *http.Request) {
}

if err != nil {
http.Error(w, fmt.Sprintf(app.UnableToReadBodyErrFormat, err), http.StatusBadRequest)
http.Error(w, app.ErrUnableToReadBody(err), http.StatusBadRequest)
return
}

Expand Down
25 changes: 23 additions & 2 deletions cmd/collector/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
package main

import (
"errors"
"fmt"
"io"
"log"
Expand Down Expand Up @@ -46,13 +47,15 @@ import (
"github.com/jaegertracing/jaeger/cmd/docs"
"github.com/jaegertracing/jaeger/cmd/env"
"github.com/jaegertracing/jaeger/cmd/flags"
clientcfgHandler "github.com/jaegertracing/jaeger/pkg/clientcfg/clientcfghttp"
"github.com/jaegertracing/jaeger/pkg/config"
"github.com/jaegertracing/jaeger/pkg/healthcheck"
"github.com/jaegertracing/jaeger/pkg/recoveryhandler"
"github.com/jaegertracing/jaeger/pkg/version"
ss "github.com/jaegertracing/jaeger/plugin/sampling/strategystore"
"github.com/jaegertracing/jaeger/plugin/storage"
"github.com/jaegertracing/jaeger/ports"
"github.com/jaegertracing/jaeger/thrift-gen/baggage"
jc "github.com/jaegertracing/jaeger/thrift-gen/jaeger"
sc "github.com/jaegertracing/jaeger/thrift-gen/sampling"
zc "github.com/jaegertracing/jaeger/thrift-gen/zipkincore"
Expand Down Expand Up @@ -133,8 +136,14 @@ func main() {

{
r := mux.NewRouter()
apiHandler := app.NewAPIHandler(jaegerBatchesHandler)
apiHandler.RegisterRoutes(r)
configManager := &collectorConfigManager{samplingStrategy: strategyStore}
cfgParams := clientcfgHandler.HTTPHandlerParams{
ConfigManager: configManager,
MetricsFactory: metricsFactory,
LegacySamplingEndpoint: false,
}
endpointHandler := app.NewAPIHandler(jaegerBatchesHandler, *clientcfgHandler.NewHTTPHandler(cfgParams))
endpointHandler.RegisterRoutes(r)
httpPortStr := ":" + strconv.Itoa(builderOpts.CollectorHTTPPort)
recoveryHandler := recoveryhandler.NewRecoveryHandler(logger, true)
httpHandler := recoveryHandler(r)
Expand Down Expand Up @@ -241,6 +250,18 @@ func startZipkinHTTPAPI(
}
}

type collectorConfigManager struct {
samplingStrategy strategystore.StrategyStore
}

func (c *collectorConfigManager) GetSamplingStrategy(serviceName string) (*sc.SamplingStrategyResponse, error) {
return c.samplingStrategy.GetSamplingStrategy(serviceName)
}

func (c *collectorConfigManager) GetBaggageRestrictions(serviceName string) ([]*baggage.BaggageRestriction, error) {
return nil, errors.New("baggage not implemented")
}

func initSamplingStrategyStore(
samplingStrategyStoreFactory *ss.Factory,
metricsFactory metrics.Factory,
Expand Down

0 comments on commit 7350001

Please sign in to comment.