Skip to content

Commit 8cd5aed

Browse files
pavolloffaypavolloffay
committed
Collect Zipkin v2 json
Signed-off-by: Pavol Loffay <ploffay@redhat.com>
1 parent e0a74f2 commit 8cd5aed

26 files changed

+2021
-37
lines changed

‎.gitmodules

+3
Original file line numberDiff line numberDiff line change
@@ -4,3 +4,6 @@
44
[submodule "jaeger-ui"]
55
path = jaeger-ui
66
url = https://github.com/uber/jaeger-ui
7+
[submodule "zipkin-api"]
8+
path = zipkin-api
9+
url = https://github.com/openzipkin/zipkin-api

‎Makefile

+12-2
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
11
PROJECT_ROOT=github.com/jaegertracing/jaeger
2-
TOP_PKGS := $(shell glide novendor | grep -v -e ./thrift-gen/... -e ./examples/... -e ./scripts/...)
2+
TOP_PKGS := $(shell glide novendor | grep -v -e ./thrift-gen/... -e swagger-gen... -e ./examples/... -e ./scripts/...)
33

44
# all .go files that don't exist in hidden directories
5-
ALL_SRC := $(shell find . -name "*.go" | grep -v -e vendor -e thrift-gen \
5+
ALL_SRC := $(shell find . -name "*.go" | grep -v -e vendor -e thrift-gen -e swagger-gen \
66
-e ".*/\..*" \
77
-e ".*/_.*" \
88
-e ".*/mocks.*")
@@ -34,6 +34,11 @@ THRIFT_GO_ARGS=thrift_import="github.com/apache/thrift/lib/go/thrift"
3434
THRIFT_GEN=$(shell which thrift-gen)
3535
THRIFT_GEN_DIR=thrift-gen
3636

37+
SWAGGER_VER=0.12.0
38+
SWAGGER_IMAGE=quay.io/goswagger/swagger:$(SWAGGER_VER)
39+
SWAGGER=docker run --rm -it -u ${shell id -u} -v "${PWD}:/go/src/${PROJECT_ROOT}" -w /go/src/${PROJECT_ROOT} $(SWAGGER_IMAGE)
40+
SWAGGER_GEN_DIR=swagger-gen
41+
3742
PASS=$(shell printf "\033[32mPASS\033[0m")
3843
FAIL=$(shell printf "\033[31mFAIL\033[0m")
3944
COLORIZE=$(SED) ''/PASS/s//$(PASS)/'' | $(SED) ''/FAIL/s//$(FAIL)/''
@@ -215,6 +220,11 @@ idl-submodule:
215220
git submodule init
216221
git submodule update
217222

223+
.PHONY: generate-zipkin-swagger
224+
generate-zipkin-swagger:
225+
$(SWAGGER) generate server -f ./idl/swagger/zipkin2-api.yaml -t $(SWAGGER_GEN_DIR) -O PostSpans --exclude-main
226+
rm $(SWAGGER_GEN_DIR)/restapi/operations/post_spans_urlbuilder.go $(SWAGGER_GEN_DIR)/restapi/server.go $(SWAGGER_GEN_DIR)/restapi/configure_zipkin.go $(SWAGGER_GEN_DIR)/models/trace.go $(SWAGGER_GEN_DIR)/models/list_of_traces.go $(SWAGGER_GEN_DIR)/models/dependency_link.go
227+
218228
.PHONY: thrift-image
219229
thrift-image:
220230
$(THRIFT) -version

‎cmd/collector/app/zipkin/fixtures/zipkin_01.json

+28
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
[
2+
{
3+
"traceId":"1",
4+
"id":"2",
5+
"parentId": "1",
6+
"name":"foo",
7+
"kind": "CLIENT",
8+
"debug": true,
9+
"shared": true,
10+
"timestamp": 1,
11+
"duration": 10,
12+
"localEndpoint":{
13+
"serviceName":"foo",
14+
"ipv4":"10.43.17.42"
15+
},
16+
"remoteEndpoint":{
17+
"serviceName":"bar",
18+
"ipv4":"10.43.17.43"
19+
},
20+
"annotations": [{
21+
"value": "foo",
22+
"timestamp": 1
23+
}],
24+
"tags": {
25+
"foo": "bar"
26+
}
27+
}
28+
]

‎cmd/collector/app/zipkin/http_handler.go

+77-10
Original file line numberDiff line numberDiff line change
@@ -17,49 +17,58 @@ package zipkin
1717
import (
1818
"compress/gzip"
1919
"fmt"
20+
"io"
2021
"io/ioutil"
2122
"net/http"
2223
"strings"
2324
"time"
2425

2526
"github.com/apache/thrift/lib/go/thrift"
27+
"github.com/go-openapi/loads"
28+
"github.com/go-openapi/strfmt"
29+
"github.com/go-openapi/swag"
2630
"github.com/gorilla/mux"
2731
tchanThrift "github.com/uber/tchannel-go/thrift"
2832

2933
"github.com/jaegertracing/jaeger/cmd/collector/app"
34+
"github.com/jaegertracing/jaeger/swagger-gen/models"
35+
"github.com/jaegertracing/jaeger/swagger-gen/restapi"
36+
"github.com/jaegertracing/jaeger/swagger-gen/restapi/operations"
3037
"github.com/jaegertracing/jaeger/thrift-gen/zipkincore"
3138
)
3239

3340
// APIHandler handles all HTTP calls to the collector
3441
type APIHandler struct {
3542
zipkinSpansHandler app.ZipkinSpansHandler
43+
zipkinV2Formats strfmt.Registry
3644
}
3745

3846
// NewAPIHandler returns a new APIHandler
3947
func NewAPIHandler(
4048
zipkinSpansHandler app.ZipkinSpansHandler,
41-
) *APIHandler {
49+
) (*APIHandler, error) {
50+
swaggerSpec, _ := loads.Analyzed(restapi.SwaggerJSON, "")
4251
return &APIHandler{
4352
zipkinSpansHandler: zipkinSpansHandler,
44-
}
53+
zipkinV2Formats: operations.NewZipkinAPI(swaggerSpec).Formats(),
54+
}, nil
4555
}
4656

4757
// RegisterRoutes registers Zipkin routes
4858
func (aH *APIHandler) RegisterRoutes(router *mux.Router) {
4959
router.HandleFunc("/api/v1/spans", aH.saveSpans).Methods(http.MethodPost)
60+
router.HandleFunc("/api/v2/spans", aH.saveSpansV2).Methods(http.MethodPost)
5061
}
5162

5263
func (aH *APIHandler) saveSpans(w http.ResponseWriter, r *http.Request) {
5364
bRead := r.Body
5465
defer r.Body.Close()
55-
5666
if strings.Contains(r.Header.Get("Content-Encoding"), "gzip") {
57-
gz, err := gzip.NewReader(r.Body)
67+
gz, err := gunzip(bRead)
5868
if err != nil {
5969
http.Error(w, fmt.Sprintf(app.UnableToReadBodyErrFormat, err), http.StatusBadRequest)
6070
return
6171
}
62-
defer gz.Close()
6372
bRead = gz
6473
}
6574

@@ -84,15 +93,73 @@ func (aH *APIHandler) saveSpans(w http.ResponseWriter, r *http.Request) {
8493
return
8594
}
8695

87-
if len(tSpans) > 0 {
88-
ctx, _ := tchanThrift.NewContext(time.Minute)
89-
if _, err = aH.zipkinSpansHandler.SubmitZipkinBatch(ctx, tSpans); err != nil {
90-
http.Error(w, fmt.Sprintf("Cannot submit Zipkin batch: %v", err), http.StatusInternalServerError)
96+
if err := aH.saveThriftSpans(tSpans); err != nil {
97+
http.Error(w, fmt.Sprintf("Cannot submit Zipkin batch: %v", err), http.StatusInternalServerError)
98+
return
99+
}
100+
101+
w.WriteHeader(http.StatusAccepted)
102+
}
103+
104+
func (aH *APIHandler) saveSpansV2(w http.ResponseWriter, r *http.Request) {
105+
bRead := r.Body
106+
defer r.Body.Close()
107+
if strings.Contains(r.Header.Get("Content-Encoding"), "gzip") {
108+
gz, err := gunzip(bRead)
109+
if err != nil {
110+
http.Error(w, fmt.Sprintf(app.UnableToReadBodyErrFormat, err), http.StatusBadRequest)
91111
return
92112
}
113+
bRead = gz
93114
}
94115

95-
w.WriteHeader(http.StatusAccepted)
116+
bodyBytes, err := ioutil.ReadAll(bRead)
117+
if err != nil {
118+
http.Error(w, fmt.Sprintf(app.UnableToReadBodyErrFormat, err), http.StatusInternalServerError)
119+
return
120+
}
121+
122+
var spans models.ListOfSpans
123+
if err = swag.ReadJSON(bodyBytes, &spans); err != nil {
124+
http.Error(w, fmt.Sprintf(app.UnableToReadBodyErrFormat, err), http.StatusBadRequest)
125+
return
126+
}
127+
if err = spans.Validate(aH.zipkinV2Formats); err != nil {
128+
http.Error(w, fmt.Sprintf(app.UnableToReadBodyErrFormat, err), http.StatusBadRequest)
129+
return
130+
}
131+
132+
tSpans, err := spansV2ToThrift(&spans)
133+
if err != nil {
134+
http.Error(w, fmt.Sprintf(app.UnableToReadBodyErrFormat, err), http.StatusBadRequest)
135+
return
136+
}
137+
138+
if err := aH.saveThriftSpans(tSpans); err != nil {
139+
http.Error(w, fmt.Sprintf("Cannot submit Zipkin batch: %v", err), http.StatusInternalServerError)
140+
return
141+
}
142+
143+
w.WriteHeader(operations.PostSpansAcceptedCode)
144+
}
145+
146+
func gunzip(r io.ReadCloser) (*gzip.Reader, error) {
147+
gz, err := gzip.NewReader(r)
148+
if err != nil {
149+
return nil, err
150+
}
151+
defer gz.Close()
152+
return gz, nil
153+
}
154+
155+
func (aH *APIHandler) saveThriftSpans(tSpans []*zipkincore.Span) error {
156+
if len(tSpans) > 0 {
157+
ctx, _ := tchanThrift.NewContext(time.Minute)
158+
if _, err := aH.zipkinSpansHandler.SubmitZipkinBatch(ctx, tSpans); err != nil {
159+
return err
160+
}
161+
}
162+
return nil
96163
}
97164

98165
func deserializeThrift(b []byte) ([]*zipkincore.Span, error) {

‎cmd/collector/app/zipkin/http_handler_test.go

+3-2
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ func (p *mockZipkinHandler) getSpans() []*zipkincore.Span {
5959

6060
func initializeTestServer(err error) (*httptest.Server, *APIHandler) {
6161
r := mux.NewRouter()
62-
handler := NewAPIHandler(&mockZipkinHandler{err: err})
62+
handler, _ := NewAPIHandler(&mockZipkinHandler{err: err})
6363
handler.RegisterRoutes(r)
6464
return httptest.NewServer(r), handler
6565
}
@@ -224,7 +224,8 @@ func TestDeserializeWithBadListStart(t *testing.T) {
224224
}
225225

226226
func TestCannotReadBodyFromRequest(t *testing.T) {
227-
handler := NewAPIHandler(&mockZipkinHandler{})
227+
handler, err := NewAPIHandler(&mockZipkinHandler{})
228+
require.NoError(t, err)
228229
req, err := http.NewRequest(http.MethodPost, "whatever", &errReader{})
229230
assert.NoError(t, err)
230231
rw := dummyResponseWriter{}

‎cmd/collector/app/zipkin/json.go

+16-10
Original file line numberDiff line numberDiff line change
@@ -153,29 +153,35 @@ func cutLongID(id string) string {
153153
}
154154

155155
func endpointToThrift(e endpoint) (*zipkincore.Endpoint, error) {
156-
ipv4, err := parseIpv4(e.IPv4)
156+
return eToThrift(e.IPv4, e.IPv6, e.Port, e.ServiceName)
157+
}
158+
159+
func eToThrift(ip4 string, ip6 string, p int32, service string) (*zipkincore.Endpoint, error) {
160+
ipv4, err := parseIpv4(ip4)
157161
if err != nil {
158162
return nil, err
159163
}
160-
port := e.Port
161-
if port >= (1 << 15) {
162-
// Zipkin.thrift defines port as i16, so values between (2^15 and 2^16-1) must be encoded as negative
163-
port = port - (1 << 16)
164-
}
165-
166-
ipv6, err := parseIpv6(e.IPv6)
164+
port := port(p)
165+
ipv6, err := parseIpv6(string(ip6))
167166
if err != nil {
168167
return nil, err
169168
}
170-
171169
return &zipkincore.Endpoint{
172-
ServiceName: e.ServiceName,
170+
ServiceName: service,
173171
Port: int16(port),
174172
Ipv4: ipv4,
175173
Ipv6: ipv6,
176174
}, nil
177175
}
178176

177+
func port(p int32) int32 {
178+
if p >= (1 << 15) {
179+
// Zipkin.thrift defines port as i16, so values between (2^15 and 2^16-1) must be encoded as negative
180+
p = p - (1 << 16)
181+
}
182+
return p
183+
}
184+
179185
func annoToThrift(a annotation) (*zipkincore.Annotation, error) {
180186
endpoint, err := endpointToThrift(a.Endpoint)
181187
if err != nil {

0 commit comments

Comments
 (0)