Skip to content

Commit 84982e3

Browse files
marcsanmimarcsanmi
committed
add support to ingest otlp in the http server & update proto version
1 parent c0045e6 commit 84982e3

File tree

7 files changed

+166
-139
lines changed

7 files changed

+166
-139
lines changed

‎api/otlp/Makefile

+1-1
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ endef
1616
OPENTELEMETRY_PROTO_SRC_DIR=.tmp/opentelemetry-proto
1717

1818
# The SHA matching the current version of the proto to use
19-
OPENTELEMETRY_PROTO_VERSION=v0.20.0
19+
OPENTELEMETRY_PROTO_VERSION=v1.3.2
2020

2121
# Find all .proto files.
2222

‎api/otlp/collector/profiles/v1experimental/profiles_service.pb.go

+3-3
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎api/otlp/profiles/v1experimental/pprofextended.pb.go

+93-93
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎go.work.sum

+17
Original file line numberDiff line numberDiff line change
@@ -1030,6 +1030,7 @@ golang.org/x/oauth2 v0.8.0/go.mod h1:yr7u4HXZRm1R1kBWqr/xKNqewf0plRYoB7sla+BCIXE
10301030
golang.org/x/oauth2 v0.16.0/go.mod h1:hqZ+0LWXsiVoZpeld6jVt06P3adbS2Uu911W1SsJv2o=
10311031
golang.org/x/oauth2 v0.17.0/go.mod h1:OzPDGQiuQMguemayvdylqddI7qcD9lnSDb+1FiwQ5HA=
10321032
golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
1033+
golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2 h1:IRJeR9r1pYWsHKTRe/IInb7lYvbBVIqOgsX/u0mbOWY=
10331034
golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE=
10341035
golang.org/x/term v0.9.0/go.mod h1:M6DEAAIenWoTxdKrOltXcmDY3rSplQUkrvaDU5FcQyo=
10351036
golang.org/x/text v0.10.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
@@ -1040,6 +1041,9 @@ golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58
10401041
golang.org/x/tools v0.17.0/go.mod h1:xsh6VxdV005rRVaS6SSAf9oiAqljS7UZUacMZ8Bnsps=
10411042
golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 h1:+cNy6SZtPcJQH3LJVLOSmiC7MMxXNOb3PU/VUEz+EhU=
10421043
golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028/go.mod h1:NDW/Ps6MPRej6fsCIbMTohpP40sJ/P/vI1MoTEGwX90=
1044+
gonum.org/v1/gonum v0.8.2 h1:CCXrcPKiGGotvnN6jfUsKk4rRqm7q09/YbKb5xCEvtM=
1045+
gonum.org/v1/netlib v0.0.0-20190313105609-8cb42192e0e0 h1:OE9mWmgKkjJyEmDAAtGMPjXu+YNeGvK9VTSHY6+Qihc=
1046+
gonum.org/v1/plot v0.0.0-20190515093506-e2840ee46a6b h1:Qh4dB5D/WpoUUp3lSod7qgoyEHbDGPUWjIbnqdqqe1k=
10431047
google.golang.org/api v0.128.0/go.mod h1:Y611qgqaE92On/7g65MQgxYul3c0rEB894kniWLY750=
10441048
google.golang.org/api v0.160.0/go.mod h1:0mu0TpK33qnydLvWqbImq2b1eQ5FHRSDCBzAxX9ZHyw=
10451049
google.golang.org/api v0.164.0/go.mod h1:2OatzO7ZDQsoS7IFf3rvsE17/TldiU3F/zxFHeqUB5o=
@@ -1051,6 +1055,7 @@ google.golang.org/genproto/googleapis/api v0.0.0-20230530153820-e85fd2cbaebc/go.
10511055
google.golang.org/genproto/googleapis/api v0.0.0-20240123012728-ef4313101c80/go.mod h1:4jWUdICTdgc3Ibxmr8nAJiiLHwQBY0UI0XZcEMaFKaA=
10521056
google.golang.org/genproto/googleapis/api v0.0.0-20240125205218-1f4bbc51befe/go.mod h1:4jWUdICTdgc3Ibxmr8nAJiiLHwQBY0UI0XZcEMaFKaA=
10531057
google.golang.org/genproto/googleapis/api v0.0.0-20240205150955-31a09d347014/go.mod h1:rbHMSEDyoYX62nRVLOCc4Qt1HbsdytAYoVwgjiOhF3I=
1058+
google.golang.org/genproto/googleapis/bytestream v0.0.0-20240304161311-37d4d3c04a78 h1:YqFWYZXim8bG9v68xU8WjTZmYKb5M5dMeSOWIp6jogI=
10541059
google.golang.org/genproto/googleapis/bytestream v0.0.0-20240304161311-37d4d3c04a78/go.mod h1:vh/N7795ftP0AkN1w8XKqN4w1OdUKXW5Eummda+ofv8=
10551060
google.golang.org/genproto/googleapis/rpc v0.0.0-20230530153820-e85fd2cbaebc/go.mod h1:66JfowdXAEgad5O9NnYcsNPLCPZJD++2L9X0PCMODrA=
10561061
google.golang.org/genproto/googleapis/rpc v0.0.0-20240116215550-a9fa1716bcac/go.mod h1:daQN87bsDqDoe316QbbvX60nMoJQa4r6Ds0ZuoAe5yA=
@@ -1064,11 +1069,23 @@ google.golang.org/grpc v1.62.0/go.mod h1:IWTG0VlJLCh1SkC58F7np9ka9mx/WNkjl4PGJai
10641069
google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
10651070
google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
10661071
google.golang.org/protobuf v1.32.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
1072+
gopkg.in/errgo.v2 v2.1.0 h1:0vLT13EuvQ0hNvakwLuFZ/jYrLp5F3kcWHXdRggjCE8=
1073+
gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc=
10671074
gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc=
1075+
gopkg.in/square/go-jose.v2 v2.6.0 h1:NGk74WTnPKBNUhNzQX7PYcTLUjoq7mzKk2OKbvwk2iI=
10681076
gopkg.in/square/go-jose.v2 v2.6.0/go.mod h1:M9dMgbHiYLoDGQrXy7OpJDJWiKiU//h+vD76mk0e1AI=
1077+
gopkg.in/telebot.v3 v3.2.1 h1:3I4LohaAyJBiivGmkfB+CiVu7QFOWkuZ4+KHgO/G3rs=
10691078
gopkg.in/telebot.v3 v3.2.1/go.mod h1:GJKwwWqp9nSkIVN51eRKU78aB5f5OnQuWdwiIZfPbko=
1079+
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
10701080
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
1081+
gotest.tools/v3 v3.0.3 h1:4AuOwCGf4lLR9u3YOe2awrHygurzhO/HeQ6laiA6Sx0=
10711082
gotest.tools/v3 v3.0.3/go.mod h1:Z7Lb0S5l+klDB31fvDQX8ss/FlKDxtlFlw3Oa8Ymbl8=
1083+
honnef.co/go/tools v0.0.1-2020.1.4 h1:UoveltGrhghAA7ePc+e+QYDHXrBps2PqFZiHkGR/xK8=
1084+
k8s.io/gengo v0.0.0-20201113003025-83324d819ded h1:JApXBKYyB7l9xx+DK7/+mFjC7A9Bt5A93FPvFD0HIFE=
10721085
k8s.io/gengo v0.0.0-20201113003025-83324d819ded/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
10731086
k8s.io/klog v1.0.0 h1:Pt+yjF5aB1xDSVbau4VsWe+dQNzA0qv1LlXdC2dF6Q8=
10741087
k8s.io/klog v1.0.0/go.mod h1:4Bi6QPql/J/LkTDqv7R/cd3hPo4k2DG6Ptcz060Ez5I=
1088+
rsc.io/binaryregexp v0.2.0 h1:HfqmD5MEmC0zvwBuF187nq9mdnXjXsSivRiXN7SmRkE=
1089+
rsc.io/pdf v0.1.1 h1:k1MczvYDUvJBe93bYd7wrZLLUEcLZAuF824/I4e5Xr4=
1090+
rsc.io/quote/v3 v3.1.0 h1:9JKUTTIUgS6kzR9mK1YuGKv6Nl+DijDNIc0ghT58FaY=
1091+
rsc.io/sampler v1.3.0 h1:7uVkIFmeBqHfdjD+gZwtXXI+RODJ2Wc4O7MPEh/QiW4=

‎pkg/api/api.go

+6-4
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99
"context"
1010
"flag"
1111
"fmt"
12+
1213
"net/http"
1314
"strings"
1415

@@ -22,8 +23,6 @@ import (
2223
"github.com/grafana/dskit/server"
2324
grpcgw "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
2425

25-
pprofileotlp "github.com/grafana/pyroscope/api/otlp/collector/profiles/v1experimental"
26-
2726
"github.com/grafana/pyroscope/public"
2827

2928
"github.com/grafana/pyroscope/api/gen/proto/go/adhocprofiles/v1/adhocprofilesv1connect"
@@ -36,6 +35,7 @@ import (
3635
"github.com/grafana/pyroscope/api/gen/proto/go/vcs/v1/vcsv1connect"
3736
"github.com/grafana/pyroscope/api/gen/proto/go/version/v1/versionv1connect"
3837
"github.com/grafana/pyroscope/api/openapiv2"
38+
pprofileotlp "github.com/grafana/pyroscope/api/otlp/collector/profiles/v1experimental"
3939
"github.com/grafana/pyroscope/pkg/adhocprofiles"
4040
"github.com/grafana/pyroscope/pkg/compactor"
4141
"github.com/grafana/pyroscope/pkg/distributor"
@@ -214,6 +214,7 @@ func (a *API) RegisterOverridesExporter(oe *exporter.OverridesExporter) {
214214
func (a *API) RegisterDistributor(d *distributor.Distributor) {
215215
pyroscopeHandler := pyroscope.NewPyroscopeIngestHandler(d, a.logger)
216216
otlpHandler := otlp.NewOTLPIngestHandler(d, a.logger)
217+
217218
a.RegisterRoute("/ingest", pyroscopeHandler, true, true, "POST")
218219
a.RegisterRoute("/pyroscope/ingest", pyroscopeHandler, true, true, "POST")
219220
pushv1connect.RegisterPusherServiceHandler(a.server.HTTP, d, a.grpcAuthMiddleware)
@@ -222,8 +223,9 @@ func (a *API) RegisterDistributor(d *distributor.Distributor) {
222223
{Desc: "Ring status", Path: "/distributor/ring"},
223224
})
224225

225-
pprofileotlp.RegisterProfilesServiceServer(a.server.GRPC, otlpHandler)
226-
// pprofileotlp.RegisterGRPCServer(a.server.GRPC, otlpHandler)
226+
pprofileotlp.RegisterProfilesServiceServer(a.server.GRPCOnHTTPServer, otlpHandler)
227+
228+
a.RegisterRoute("/opentelemetry.proto.collector.profiles.v1experimental.ProfilesService/Export", otlpHandler, true, true, "POST")
227229

228230
// TODO(@petethepig): implement http/protobuf and http/json support
229231
// a.RegisterRoute("/v1/profiles", otlpHandler, true, true, "POST")

‎pkg/ingester/otlp/convert.go

+20-18
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,7 @@
11
package otlp
22

33
import (
4-
"encoding/json"
54
"fmt"
6-
"os"
75
"time"
86

97
googleProfile "github.com/grafana/pyroscope/api/gen/proto/go/google/v1"
@@ -51,17 +49,8 @@ func ConvertOtelToGoogle(src *otelProfile.Profile) *googleProfile.Profile {
5149
}
5250

5351
functionOffset := uint64(len(dst.Function)) + 1
54-
for _, m := range src.Mapping {
55-
dst.StringTable = append(dst.StringTable, fmt.Sprintf("%s", dst.StringTable[m.Filename]))
56-
// i == 0 function_id = functionOffset
57-
id := uint64(len(dst.Function)) + 1
58-
dst.Function = append(dst.Function, &googleProfile.Function{
59-
Id: id,
60-
Name: int64(len(dst.StringTable) - 1),
61-
})
62-
}
63-
6452
dst.Location = []*googleProfile.Location{}
53+
locationMappingIndexAddressMap := make(map[uint64]uint64)
6554
// Convert locations and mappings
6655
for i, loc := range src.Location {
6756
gl := convertLocationBack(loc)
@@ -72,6 +61,19 @@ func ConvertOtelToGoogle(src *otelProfile.Profile) *googleProfile.Profile {
7261
})
7362
}
7463
dst.Location = append(dst.Location, gl)
64+
locationMappingIndexAddressMap[loc.MappingIndex] = loc.Address
65+
}
66+
67+
for _, m := range src.Mapping {
68+
address, _ := locationMappingIndexAddressMap[m.Id]
69+
addressStr := fmt.Sprintf("%s 0x%x", dst.StringTable[m.Filename], address)
70+
dst.StringTable = append(dst.StringTable, addressStr)
71+
// i == 0 function_id = functionOffset
72+
id := uint64(len(dst.Function)) + 1
73+
dst.Function = append(dst.Function, &googleProfile.Function{
74+
Id: id,
75+
Name: int64(len(dst.StringTable) - 1),
76+
})
7577
}
7678

7779
// Convert samples
@@ -90,12 +92,12 @@ func ConvertOtelToGoogle(src *otelProfile.Profile) *googleProfile.Profile {
9092
dst.DefaultSampleType = int64(len(dst.StringTable) - 2)
9193
}
9294

93-
b, _ := json.MarshalIndent(src, "", " ")
94-
fmt.Println("src:")
95-
_, _ = fmt.Fprintln(os.Stdout, string(b))
96-
b, _ = json.MarshalIndent(dst, "", " ")
97-
fmt.Println("dst:")
98-
_, _ = fmt.Fprintln(os.Stdout, string(b))
95+
//b, _ := json.MarshalIndent(src, "", " ")
96+
//fmt.Println("src:")
97+
//_, _ = fmt.Fprintln(os.Stdout, string(b))
98+
//b, _ = json.MarshalIndent(dst, "", " ")
99+
//fmt.Println("dst:")
100+
//_, _ = fmt.Fprintln(os.Stdout, string(b))
99101

100102
return dst
101103
}

‎pkg/ingester/otlp/ingest_handler.go

+26-20
Original file line numberDiff line numberDiff line change
@@ -5,23 +5,24 @@ import (
55
"fmt"
66
"net/http"
77
"os"
8+
"strings"
89

910
"connectrpc.com/connect"
1011
"github.com/go-kit/log"
11-
"github.com/go-kit/log/level"
1212
"github.com/google/uuid"
13+
"google.golang.org/grpc"
1314

1415
pushv1 "github.com/grafana/pyroscope/api/gen/proto/go/push/v1"
1516
typesv1 "github.com/grafana/pyroscope/api/gen/proto/go/types/v1"
16-
"github.com/grafana/pyroscope/pkg/tenant"
17-
1817
pprofileotlp "github.com/grafana/pyroscope/api/otlp/collector/profiles/v1experimental"
18+
"github.com/grafana/pyroscope/pkg/tenant"
1919
)
2020

2121
type ingestHandler struct {
2222
pprofileotlp.UnimplementedProfilesServiceServer
23-
svc PushService
24-
log log.Logger
23+
svc PushService
24+
log log.Logger
25+
handler http.Handler
2526
}
2627

2728
// TODO(@petethepig): split http and grpc
@@ -35,26 +36,31 @@ type PushService interface {
3536
}
3637

3738
func NewOTLPIngestHandler(svc PushService, l log.Logger) Handler {
38-
return &ingestHandler{
39+
h := &ingestHandler{
3940
svc: svc,
40-
log: level.Error(l),
41+
log: l,
4142
}
43+
44+
grpcServer := grpc.NewServer()
45+
pprofileotlp.RegisterProfilesServiceServer(grpcServer, h)
46+
47+
h.handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
48+
if r.ProtoMajor == 2 && strings.HasPrefix(r.Header.Get("Content-Type"), "application/grpc") {
49+
grpcServer.ServeHTTP(w, r)
50+
return
51+
}
52+
53+
// Handle HTTP requests (if we want to support HTTP/Protobuf in future)
54+
//if r.URL.Path == "/v1/profiles" {}
55+
56+
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
57+
})
58+
59+
return h
4260
}
4361

44-
// TODO(@petethepig): implement
45-
// TODO(@petethepig): split http and grpc
4662
func (h *ingestHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
47-
panic("not implemented")
48-
49-
req := &pushv1.PushRequest{}
50-
_, err := h.svc.Push(r.Context(), connect.NewRequest(req))
51-
if err != nil {
52-
w.WriteHeader(http.StatusInternalServerError)
53-
w.Write([]byte("Internal Server Error: " + err.Error()))
54-
return
55-
}
56-
w.WriteHeader(http.StatusOK)
57-
w.Write([]byte("OK"))
63+
h.handler.ServeHTTP(w, r)
5864
}
5965

6066
// TODO(@petethepig): split http and grpc

0 commit comments

Comments
 (0)