[feat] introduce data stream client

Signed-off-by: zzzk1 <[email protected]>

pkg/es/client/datastream_client.go

@@ -0,0 +1,33 @@
+// Copyright (c) 2025 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package client
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+)
+
+var _ DataStreamAPI = (*DataStreamClient)(nil)
+
+type DataStreamClient struct {
+	Client
+}
+
+func (ds DataStreamClient) Create(dataStream string) error {
+	_, err := ds.request(elasticRequest{
+		endpoint: "_data_stream/" + dataStream,
+		method:   http.MethodPut,
+	})
+	if err != nil {
+		var responseError ResponseError
+		if errors.As(err, &responseError) {
+			if responseError.StatusCode != http.StatusOK {
+				return responseError.prefixMessage("failed to create data stream: " + dataStream)
+			}
+		}
+		return fmt.Errorf("failed to create data stream: %w", err)
+	}
+	return nil
+}

pkg/es/client/datastream_client_test.go

@@ -0,0 +1,60 @@
+// Copyright (c) 2025 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package client
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCreate(t *testing.T) {
+	dataStreamName := "jaeger-span"
+	tests := []struct {
+		name         string
+		responseCode int
+		response     string
+		errContains  string
+	}{
+		{
+			name:         "success",
+			responseCode: http.StatusOK,
+		},
+		{
+			name:         "client error",
+			responseCode: http.StatusBadRequest,
+			response:     esErrResponse,
+			errContains:  "failed to create data stream: jaeger-span",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			testServer := httptest.NewServer(http.HandlerFunc(func(res http.ResponseWriter, req *http.Request) {
+				assert.True(t, strings.HasSuffix(req.URL.String(), dataStreamName))
+				assert.Equal(t, http.MethodPut, req.Method)
+				assert.Equal(t, "Basic foobar", req.Header.Get("Authorization"))
+				res.WriteHeader(test.responseCode)
+				res.Write([]byte(test.response))
+			}))
+			defer testServer.Close()
+
+			c := &DataStreamClient{
+				Client: Client{
+					Client:    testServer.Client(),
+					Endpoint:  testServer.URL,
+					BasicAuth: "foobar",
+				},
+			}
+			err := c.Create(dataStreamName)
+			if test.errContains != "" {
+				require.ErrorContains(t, err, test.errContains)
+			}
+		})
+	}
+}

pkg/es/client/ilm_client_test.go

@@ -68,7 +68,7 @@ func TestExists(t *testing.T) {
 	}
 }
 
-const esILMPolicy = `
+const defaultILMPolicy = `
 {
   "policy": {
     "phases": {
@@ -94,7 +94,7 @@ const esILMPolicy = `
 }
 `
 
-func TestCreate(t *testing.T) {
+func TestCreateOrUpdate(t *testing.T) {
 	policy := "jaeger-ilm-policy"
 	tests := []struct {
 		name         string
@@ -106,6 +106,11 @@ func TestCreate(t *testing.T) {
 			name:         "successful",
 			responseCode: http.StatusOK,
 		},
+		{
+			name:         "not found",
+			responseCode: http.StatusNotFound,
+			response:     esErrResponse,
+		},
 		{
 			name:         "client error",
 			responseCode: http.StatusBadRequest,
@@ -131,7 +136,7 @@ func TestCreate(t *testing.T) {
 				},
 			}
 			ilmPolicy := types.NewIlmPolicy()
-			json.Unmarshal([]byte(esILMPolicy), ilmPolicy)
+			json.Unmarshal([]byte(defaultILMPolicy), ilmPolicy)
 			fmt.Printf("%+v\n", *ilmPolicy)
 
 			err := c.CreateOrUpdate(policy, Policy{ILMPolicy: *ilmPolicy})

pkg/es/client/interfaces.go

@@ -21,3 +21,7 @@ type IndexManagementLifecycleAPI interface {
 	Exists(name string) (bool, error)
 	CreateOrUpdate(policy string, ilmPolicy Policy) error
 }
+
+type DataStreamAPI interface {
+	Create(name string) error
+}