Streams support POC

generated/kbapi/Makefile

@@ -2,7 +2,7 @@
 SHELL := /bin/bash
 ROOT_DIR := $(shell dirname $(realpath $(firstword $(MAKEFILE_LIST))))
 
-github_ref ?= 6a867d8e589056123e9e08e61256f48f6edb092e
+github_ref ?= 660b9ea6419cc5d532ce34697e55109a109fbe02
 oas_url    := https://raw.githubusercontent.com/elastic/kibana/$(github_ref)/oas_docs/output/kibana.yaml
 
 .PHONY: all

internal/clients/fleet/fleet.go

@@ -343,7 +343,7 @@ func CreatePackagePolicy(ctx context.Context, client *Client, spaceID string, re
 // UpdatePackagePolicy updates an existing package policy.
 func UpdatePackagePolicy(ctx context.Context, client *Client, id string, spaceID string, req kbapi.PackagePolicyRequest) (*kbapi.PackagePolicy, diag.Diagnostics) {
 	params := kbapi.PutFleetPackagePoliciesPackagepolicyidParams{
-		Format: utils.Pointer(kbapi.Simplified),
+		Format: utils.Pointer(kbapi.PutFleetPackagePoliciesPackagepolicyidParamsFormatSimplified),
 	}
 
 	resp, err := client.API.PutFleetPackagePoliciesPackagepolicyidWithResponse(ctx, id, &params, req, spaceAwarePathRequestEditor(spaceID))

internal/clients/kibana_oapi/streams.go

@@ -0,0 +1,193 @@
+package kibana_oapi
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"net/http"
+	"strings"
+
+	"github.com/elastic/terraform-provider-elasticstack/generated/kbapi"
+	"github.com/elastic/terraform-provider-elasticstack/internal/diagutil"
+	"github.com/hashicorp/terraform-plugin-framework/diag"
+)
+
+func GetStreamJSON(ctx context.Context, client *Client, name string) ([]byte, diag.Diagnostics) {
+	resp, err := client.API.GetStreamsNameWithResponse(ctx, name)
+	if err != nil {
+		return nil, diagutil.FrameworkDiagFromError(err)
+	}
+
+	status := resp.StatusCode()
+	switch {
+	case status == http.StatusNotFound:
+		return nil, nil
+	case status >= 200 && status < 300:
+		return resp.Body, nil
+	default:
+		return nil, reportUnknownError(status, resp.Body)
+	}
+}
+
+func PutStreamRaw(ctx context.Context, client *Client, name string, body []byte) diag.Diagnostics {
+	var diags diag.Diagnostics
+
+	endpoint := client.URL
+	if !strings.HasSuffix(endpoint, "/") {
+		endpoint += "/"
+	}
+
+	url := endpoint + "api/streams/" + name
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPut, url, bytes.NewReader(body))
+	if err != nil {
+		return diagutil.FrameworkDiagFromError(err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := client.HTTP.Do(req)
+	if err != nil {
+		return diagutil.FrameworkDiagFromError(err)
+	}
+	defer resp.Body.Close()
+
+	respBody, _ := io.ReadAll(resp.Body)
+
+	status := resp.StatusCode
+	if status >= 200 && status < 300 {
+		return diags
+	}
+
+	return reportUnknownError(status, respBody)
+}
+
+func DeleteStream(ctx context.Context, client *Client, name string) diag.Diagnostics {
+	var diags diag.Diagnostics
+
+	endpoint := client.URL
+	if !strings.HasSuffix(endpoint, "/") {
+		endpoint += "/"
+	}
+
+	url := endpoint + "api/streams/" + name
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodDelete, url, nil)
+	if err != nil {
+		return diagutil.FrameworkDiagFromError(err)
+	}
+
+	resp, err := client.HTTP.Do(req)
+	if err != nil {
+		return diagutil.FrameworkDiagFromError(err)
+	}
+	defer resp.Body.Close()
+
+	respBody, _ := io.ReadAll(resp.Body)
+
+	switch resp.StatusCode {
+	case http.StatusNotFound:
+		return diags
+	case http.StatusOK, http.StatusNoContent:
+		return diags
+	case http.StatusBadRequest:
+		if bytes.Contains(respBody, []byte("Expected undefined, received null")) {
+			return diags
+		}
+		return reportUnknownError(resp.StatusCode, respBody)
+	default:
+		return reportUnknownError(resp.StatusCode, respBody)
+	}
+}
+
+func GetStreamIngestJSON(ctx context.Context, client *Client, name string) ([]byte, diag.Diagnostics) {
+	resp, err := client.API.GetStreamsNameIngestWithResponse(ctx, name)
+	if err != nil {
+		return nil, diagutil.FrameworkDiagFromError(err)
+	}
+
+	status := resp.StatusCode()
+	switch {
+	case status == http.StatusNotFound:
+		return nil, nil
+	case status == http.StatusBadRequest:
+
+		return nil, nil
+	case status >= 200 && status < 300:
+		return resp.Body, nil
+	default:
+		return nil, reportUnknownError(status, resp.Body)
+	}
+}
+
+func PutStreamIngest(ctx context.Context, client *Client, name string, body kbapi.PutStreamsNameIngestJSONRequestBody) diag.Diagnostics {
+	resp, err := client.API.PutStreamsNameIngestWithResponse(ctx, name, body)
+	if err != nil {
+		return diagutil.FrameworkDiagFromError(err)
+	}
+
+	status := resp.StatusCode()
+	if status >= 200 && status < 300 {
+		return nil
+	}
+
+	return reportUnknownError(status, resp.Body)
+}
+
+func GetStreamGroupJSON(ctx context.Context, client *Client, name string) ([]byte, diag.Diagnostics) {
+	resp, err := client.API.GetStreamsNameGroupWithResponse(ctx, name)
+	if err != nil {
+		return nil, diagutil.FrameworkDiagFromError(err)
+	}
+
+	status := resp.StatusCode()
+	switch {
+	case status == http.StatusNotFound:
+		return nil, nil
+	case status >= 200 && status < 300:
+		return resp.Body, nil
+	default:
+		return nil, reportUnknownError(status, resp.Body)
+	}
+}
+
+func PutStreamGroup(ctx context.Context, client *Client, name string, body kbapi.PutStreamsNameGroupJSONRequestBody) diag.Diagnostics {
+	resp, err := client.API.PutStreamsNameGroupWithResponse(ctx, name, body)
+	if err != nil {
+		return diagutil.FrameworkDiagFromError(err)
+	}
+
+	status := resp.StatusCode()
+	if status >= 200 && status < 300 {
+		return nil
+	}
+
+	return reportUnknownError(status, resp.Body)
+}
+
+func EnableStreams(ctx context.Context, client *Client) diag.Diagnostics {
+	resp, err := client.API.PostStreamsEnableWithResponse(ctx)
+	if err != nil {
+		return diagutil.FrameworkDiagFromError(err)
+	}
+
+	status := resp.StatusCode()
+	if status >= 200 && status < 300 {
+		return nil
+	}
+
+	return reportUnknownError(status, resp.Body)
+}
+
+func DisableStreams(ctx context.Context, client *Client) diag.Diagnostics {
+	resp, err := client.API.PostStreamsDisableWithResponse(ctx)
+	if err != nil {
+		return diagutil.FrameworkDiagFromError(err)
+	}
+
+	status := resp.StatusCode()
+	if status >= 200 && status < 300 {
+		return nil
+	}
+
+	return reportUnknownError(status, resp.Body)
+}

internal/kibana/streams/conditions.go

@@ -0,0 +1,65 @@
+package streams
+
+import "encoding/json"
+
+// Condition represents a logical condition tree that can be serialized
+// into a JSON-friendly DSL structure.
+
+type Condition interface {
+	ToDSL() map[string]any
+}
+
+type FieldComparison struct {
+	Field string
+	Op    string
+	Value any
+}
+
+func (c FieldComparison) ToDSL() map[string]any {
+	return map[string]any{
+		"field": c.Field,
+		"op":    c.Op,
+		"value": c.Value,
+	}
+}
+
+type And struct {
+	Children []Condition
+}
+
+func (a And) ToDSL() map[string]any {
+	children := make([]any, 0, len(a.Children))
+	for _, child := range a.Children {
+		if child == nil {
+			continue
+		}
+		children = append(children, child.ToDSL())
+	}
+	return map[string]any{
+		"and": children,
+	}
+}
+
+type Or struct {
+	Children []Condition
+}
+
+func (o Or) ToDSL() map[string]any {
+	children := make([]any, 0, len(o.Children))
+	for _, child := range o.Children {
+		if child == nil {
+			continue
+		}
+		children = append(children, child.ToDSL())
+	}
+	return map[string]any{
+		"or": children,
+	}
+}
+
+func MarshalCondition(c Condition) ([]byte, error) {
+	if c == nil {
+		return []byte("null"), nil
+	}
+	return json.Marshal(c.ToDSL())
+}

internal/kibana/streams/conditions_test.go

@@ -0,0 +1,99 @@
+package streams
+
+import (
+	"encoding/json"
+	"reflect"
+	"testing"
+)
+
+func mustDecodeJSON(t *testing.T, b []byte) any {
+	t.Helper()
+	var v any
+	if err := json.Unmarshal(b, &v); err != nil {
+		t.Fatalf("failed to unmarshal JSON: %v\ninput: %s", err, string(b))
+	}
+	return v
+}
+
+func TestMarshalCondition_SimpleAndOr(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		cond Condition
+		want string
+	}{
+		{
+			name: "single leaf",
+			cond: FieldComparison{
+				Field: "host.name",
+				Op:    "eq",
+				Value: "web-01",
+			},
+			want: `{
+				"field": "host.name",
+				"op":    "eq",
+				"value": "web-01"
+			}`,
+		},
+		{
+			name: "and of two leaves",
+			cond: And{
+				Children: []Condition{
+					FieldComparison{Field: "host.name", Op: "eq", Value: "web-01"},
+					FieldComparison{Field: "status", Op: "eq", Value: "ok"},
+				},
+			},
+			want: `{
+				"and": [
+					{"field": "host.name", "op": "eq", "value": "web-01"},
+					{"field": "status",    "op": "eq", "value": "ok"}
+				]
+			}`,
+		},
+		{
+			name: "nested and/or tree",
+			cond: And{
+				Children: []Condition{
+					FieldComparison{Field: "env", Op: "eq", Value: "prod"},
+					Or{
+						Children: []Condition{
+							FieldComparison{Field: "service", Op: "eq", Value: "api"},
+							FieldComparison{Field: "service", Op: "eq", Value: "frontend"},
+						},
+					},
+				},
+			},
+			want: `{
+				"and": [
+					{"field": "env", "op": "eq", "value": "prod"},
+					{"or": [
+						{"field": "service", "op": "eq", "value": "api"},
+						{"field": "service", "op": "eq", "value": "frontend"}
+					]}
+				]
+			}`,
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			gotBytes, err := MarshalCondition(tt.cond)
+			if err != nil {
+				t.Fatalf("MarshalCondition() error = %v", err)
+			}
+
+			got := mustDecodeJSON(t, gotBytes)
+			want := mustDecodeJSON(t, []byte(tt.want))
+
+			if !reflect.DeepEqual(got, want) {
+				gb, _ := json.MarshalIndent(got, "", "  ")
+				wb, _ := json.MarshalIndent(want, "", "  ")
+				t.Fatalf("mismatch.\nGot:\n%s\nWant:\n%s", string(gb), string(wb))
+			}
+		})
+	}
+}