enhance: Support otlp http exporter (#35053)

See also #35052

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>

configs/milvus.yaml

@@ -850,7 +850,8 @@ trace:
   jaeger:
     url:  # when exporter is jaeger should set the jaeger's URL
   otlp:
-    endpoint:  # example: "127.0.0.1:4318"
+    endpoint:  # example: "127.0.0.1:4317" for grpc, "127.0.0.1:4318" for http
+    method:  # otlp export method, acceptable values: ["grpc", "http"],  using "grpc" by default
     secure: true
   initTimeoutSeconds: 10 # segcore initialization timeout in seconds, preventing otlp grpc hangs forever
 
@@ -862,4 +863,4 @@ trace:
 #maxMemSize will the whole available GPU memory.
 gpu:
   initMemSize:  # Gpu Memory Pool init size
-  maxMemSize:  # Gpu Memory Pool Max size
+  maxMemSize:  # Gpu Memory Pool Max size

go.mod

@@ -219,6 +219,7 @@ require (
 	go.opentelemetry.io/otel/exporters/jaeger v1.13.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.20.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.20.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.20.0 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.20.0 // indirect
 	go.opentelemetry.io/otel/metric v1.20.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.20.0 // indirect

go.sum

@@ -966,6 +966,8 @@ go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.20.0/go.mod h1:GijYcYmNpX1K
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.0.1/go.mod h1:xOvWoTOrQjxjW61xtOmD/WKGRYb/P4NzRo3bs65U6Rk=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.20.0 h1:gvmNvqrPYovvyRmCSygkUDyL8lC5Tl845MLEwqpxhEU=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.20.0/go.mod h1:vNUq47TGFioo+ffTSnKNdob241vePmtNZnAODKapKd0=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.20.0 h1:CsBiKCiQPdSjS+MlRiqeTI9JDDpSuk0Hb6QTRfwer8k=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.20.0/go.mod h1:CMJYNAfooOwSZSAmAeMUV1M+TXld3BiK++z9fqIm2xk=
 go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.20.0 h1:4s9HxB4azeeQkhY0GE5wZlMj4/pz8tE5gx2OQpGUw58=
 go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.20.0/go.mod h1:djVA3TUJ2fSdMX0JE5XxFBOaZzprElJoP7fD4vnV2SU=
 go.opentelemetry.io/otel/metric v1.20.0 h1:ZlrO8Hu9+GAhnepmRGhSU7/VkpjrNowxRN9GyKR4wzA=

internal/core/src/common/Tracer.cpp

@@ -10,6 +10,8 @@
 // or implied. See the License for the specific language governing permissions and limitations under the License
 
 #include "Tracer.h"
+#include <opentelemetry/exporters/otlp/otlp_http_exporter_factory.h>
+#include <opentelemetry/exporters/otlp/otlp_http_exporter_options.h>
 #include "log/Log.h"
 
 #include <iomanip>
@@ -57,11 +59,22 @@ initTelemetry(const TraceConfig& cfg) {
         exporter = jaeger::JaegerExporterFactory::Create(opts);
         LOG_INFO("init jaeger exporter, endpoint: {}", opts.endpoint);
     } else if (cfg.exporter == "otlp") {
-        auto opts = otlp::OtlpGrpcExporterOptions{};
-        opts.endpoint = cfg.otlpEndpoint;
-        opts.use_ssl_credentials = cfg.oltpSecure;
-        exporter = otlp::OtlpGrpcExporterFactory::Create(opts);
-        LOG_INFO("init otlp exporter, endpoint: {}", opts.endpoint);
+        if (cfg.otlpMethod == "http") {
+            auto opts = otlp::OtlpHttpExporterOptions{};
+            opts.url = cfg.otlpEndpoint;
+            exporter = otlp::OtlpHttpExporterFactory::Create(opts);
+            LOG_INFO("init otlp http exporter, endpoint: {}", opts.url);
+        } else if (cfg.otlpMethod == "grpc" || 
+            cfg.otlpMethod == "") { // legacy configuration
+            auto opts = otlp::OtlpGrpcExporterOptions{};
+            opts.endpoint = cfg.otlpEndpoint;
+            opts.use_ssl_credentials = cfg.oltpSecure;
+            exporter = otlp::OtlpGrpcExporterFactory::Create(opts);
+            LOG_INFO("init otlp grpc exporter, endpoint: {}", opts.endpoint);
+        } else {
+            LOG_INFO("unknown otlp exporter method: {}", cfg.otlpMethod);
+            enable_trace = false;
+        }
     } else {
         LOG_INFO("Empty Trace");
         enable_trace = false;

internal/core/src/common/Tracer.h

@@ -25,6 +25,7 @@ struct TraceConfig {
     float sampleFraction;
     std::string jaegerURL;
     std::string otlpEndpoint;
+    std::string otlpMethod;
     bool oltpSecure;
 
     int nodeID;

internal/core/src/common/init_c.cpp

@@ -84,6 +84,7 @@ InitTrace(CTraceConfig* config) {
                                                    config->sampleFraction,
                                                    config->jaegerURL,
                                                    config->otlpEndpoint,
+                                                   config->otlpMethod,
                                                    config->oltpSecure,
                                                    config->nodeID};
     std::call_once(
@@ -100,6 +101,7 @@ SetTrace(CTraceConfig* config) {
                                                    config->sampleFraction,
                                                    config->jaegerURL,
                                                    config->otlpEndpoint,
+                                                   config->otlpMethod,
                                                    config->oltpSecure,
                                                    config->nodeID};
     milvus::tracer::initTelemetry(traceConfig);

internal/core/src/common/type_c.h

@@ -107,6 +107,7 @@ typedef struct CTraceConfig {
     float sampleFraction;
     const char* jaegerURL;
     const char* otlpEndpoint;
+    const char* otlpMethod;
     bool oltpSecure;
 
     int nodeID;

internal/util/initcore/init_core.go

@@ -51,17 +51,20 @@ func InitTraceConfig(params *paramtable.ComponentParam) {
 	nodeID := C.int(paramtable.GetNodeID())
 	exporter := C.CString(params.TraceCfg.Exporter.GetValue())
 	jaegerURL := C.CString(params.TraceCfg.JaegerURL.GetValue())
+	otlpMethod := C.CString(params.TraceCfg.OtlpMethod.GetValue())
 	endpoint := C.CString(params.TraceCfg.OtlpEndpoint.GetValue())
 	otlpSecure := params.TraceCfg.OtlpSecure.GetAsBool()
 	defer C.free(unsafe.Pointer(exporter))
 	defer C.free(unsafe.Pointer(jaegerURL))
 	defer C.free(unsafe.Pointer(endpoint))
+	defer C.free(unsafe.Pointer(otlpMethod))
 
 	config := C.CTraceConfig{
 		exporter:       exporter,
 		sampleFraction: sampleFraction,
 		jaegerURL:      jaegerURL,
 		otlpEndpoint:   endpoint,
+		otlpMethod:     otlpMethod,
 		oltpSecure:     (C.bool)(otlpSecure),
 		nodeID:         nodeID,
 	}
@@ -80,16 +83,19 @@ func ResetTraceConfig(params *paramtable.ComponentParam) {
 	exporter := C.CString(params.TraceCfg.Exporter.GetValue())
 	jaegerURL := C.CString(params.TraceCfg.JaegerURL.GetValue())
 	endpoint := C.CString(params.TraceCfg.OtlpEndpoint.GetValue())
+	otlpMethod := C.CString(params.TraceCfg.OtlpMethod.GetValue())
 	otlpSecure := params.TraceCfg.OtlpSecure.GetAsBool()
 	defer C.free(unsafe.Pointer(exporter))
 	defer C.free(unsafe.Pointer(jaegerURL))
 	defer C.free(unsafe.Pointer(endpoint))
+	defer C.free(unsafe.Pointer(otlpMethod))
 
 	config := C.CTraceConfig{
 		exporter:       exporter,
 		sampleFraction: sampleFraction,
 		jaegerURL:      jaegerURL,
 		otlpEndpoint:   endpoint,
+		otlpMethod:     otlpMethod,
 		oltpSecure:     (C.bool)(otlpSecure),
 		nodeID:         nodeID,
 	}

pkg/go.mod

@@ -41,6 +41,7 @@ require (
 	go.opentelemetry.io/otel v1.20.0
 	go.opentelemetry.io/otel/exporters/jaeger v1.13.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.20.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.20.0
 	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.20.0
 	go.opentelemetry.io/otel/sdk v1.20.0
 	go.opentelemetry.io/otel/trace v1.20.0

pkg/go.sum

@@ -820,6 +820,8 @@ go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.20.0/go.mod h1:GijYcYmNpX1K
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.0.1/go.mod h1:xOvWoTOrQjxjW61xtOmD/WKGRYb/P4NzRo3bs65U6Rk=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.20.0 h1:gvmNvqrPYovvyRmCSygkUDyL8lC5Tl845MLEwqpxhEU=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.20.0/go.mod h1:vNUq47TGFioo+ffTSnKNdob241vePmtNZnAODKapKd0=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.20.0 h1:CsBiKCiQPdSjS+MlRiqeTI9JDDpSuk0Hb6QTRfwer8k=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.20.0/go.mod h1:CMJYNAfooOwSZSAmAeMUV1M+TXld3BiK++z9fqIm2xk=
 go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.20.0 h1:4s9HxB4azeeQkhY0GE5wZlMj4/pz8tE5gx2OQpGUw58=
 go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.20.0/go.mod h1:djVA3TUJ2fSdMX0JE5XxFBOaZzprElJoP7fD4vnV2SU=
 go.opentelemetry.io/otel/metric v1.20.0 h1:ZlrO8Hu9+GAhnepmRGhSU7/VkpjrNowxRN9GyKR4wzA=

pkg/tracer/tracer.go

@@ -24,6 +24,7 @@ import (
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/exporters/jaeger"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
 	stdout "go.opentelemetry.io/otel/exporters/stdout/stdouttrace"
 	"go.opentelemetry.io/otel/propagation"
 	"go.opentelemetry.io/otel/sdk/resource"
@@ -86,13 +87,26 @@ func CreateTracerExporter(params *paramtable.ComponentParam) (sdk.SpanExporter,
 			jaeger.WithEndpoint(params.TraceCfg.JaegerURL.GetValue())))
 	case "otlp":
 		secure := params.TraceCfg.OtlpSecure.GetAsBool()
-		opts := []otlptracegrpc.Option{
-			otlptracegrpc.WithEndpoint(params.TraceCfg.OtlpEndpoint.GetValue()),
+		switch params.TraceCfg.OtlpMethod.GetValue() {
+		case "", "grpc":
+			opts := []otlptracegrpc.Option{
+				otlptracegrpc.WithEndpoint(params.TraceCfg.OtlpEndpoint.GetValue()),
+			}
+			if !secure {
+				opts = append(opts, otlptracegrpc.WithInsecure())
+			}
+			exp, err = otlptracegrpc.New(context.Background(), opts...)
+		case "http":
+			opts := []otlptracehttp.Option{
+				otlptracehttp.WithEndpoint(params.TraceCfg.OtlpEndpoint.GetValue()),
+			}
+			if !secure {
+				opts = append(opts, otlptracehttp.WithInsecure())
+			}
+			exp, err = otlptracehttp.New(context.Background(), opts...)
+		default:
+			return nil, errors.Newf("otlp method not supported: %s", params.TraceCfg.OtlpMethod.GetValue())
 		}
-		if !secure {
-			opts = append(opts, otlptracegrpc.WithInsecure())
-		}
-		exp, err = otlptracegrpc.New(context.Background(), opts...)
 	case "stdout":
 		exp, err = stdout.New()
 	case "noop":

pkg/util/paramtable/component_param.go

@@ -903,6 +903,7 @@ type traceConfig struct {
 	SampleFraction     ParamItem `refreshable:"false"`
 	JaegerURL          ParamItem `refreshable:"false"`
 	OtlpEndpoint       ParamItem `refreshable:"false"`
+	OtlpMethod         ParamItem `refreshable:"false"`
 	OtlpSecure         ParamItem `refreshable:"false"`
 	InitTimeoutSeconds ParamItem `refreshable:"false"`
 }
@@ -940,11 +941,20 @@ Fractions >= 1 will always sample. Fractions < 0 are treated as zero.`,
 	t.OtlpEndpoint = ParamItem{
 		Key:     "trace.otlp.endpoint",
 		Version: "2.3.0",
-		Doc:     "example: \"127.0.0.1:4318\"",
+		Doc:     `example: "127.0.0.1:4317" for grpc, "127.0.0.1:4318" for http`,
 		Export:  true,
 	}
 	t.OtlpEndpoint.Init(base.mgr)
 
+	t.OtlpMethod = ParamItem{
+		Key:          "trace.otlp.method",
+		Version:      "2.4.7",
+		DefaultValue: "",
+		Doc:          `otlp export method, acceptable values: ["grpc", "http"],  using "grpc" by default`,
+		Export:       true,
+	}
+	t.OtlpMethod.Init(base.mgr)
+
 	t.OtlpSecure = ParamItem{
 		Key:          "trace.otlp.secure",
 		Version:      "2.4.0",