feat: add flink module

cli/serve.go

@@ -15,6 +15,7 @@ import (
 	"github.com/goto/entropy/internal/store/postgres"
 	"github.com/goto/entropy/modules"
 	"github.com/goto/entropy/modules/firehose"
+	"github.com/goto/entropy/modules/flink"
 	"github.com/goto/entropy/modules/job"
 	"github.com/goto/entropy/modules/kafka"
 	"github.com/goto/entropy/modules/kubernetes"
@@ -92,6 +93,7 @@ func setupRegistry() module.Registry {
 		firehose.Module,
 		job.Module,
 		kafka.Module,
+		flink.Module,
 	}
 
 	registry := &modules.Registry{}

modules/flink/config.go

@@ -0,0 +1,52 @@
+package flink
+
+import (
+	_ "embed"
+	"encoding/json"
+
+	"github.com/goto/entropy/core/resource"
+	"github.com/goto/entropy/pkg/errors"
+	"github.com/goto/entropy/pkg/validator"
+)
+
+var (
+	//go:embed schema/config.json
+	configSchemaRaw []byte
+
+	validateConfig = validator.FromJSONSchema(configSchemaRaw)
+)
+
+type Influx struct {
+	URL      string `json:"url,omitempty"`
+	Username string `json:"username,omitempty"`
+	Password string `json:"password,omitempty"`
+}
+
+type Config struct {
+	KubeNamespace   string `json:"kube_namespace,omitempty"`
+	Influx          Influx `json:"influx,omitempty"`
+	SinkKafkaStream string `json:"sink_kafka_stream,omitempty"`
+}
+
+func readConfig(_ resource.Resource, confJSON json.RawMessage, dc driverConf) (*Config, error) {
+	var cfg Config
+	if err := json.Unmarshal(confJSON, &cfg); err != nil {
+		return nil, errors.ErrInvalid.WithMsgf("invalid config json").WithCausef(err.Error())
+	}
+
+	if cfg.Influx.URL == "" {
+		cfg.Influx.URL = dc.Influx.URL
+		cfg.Influx.Username = dc.Influx.Username
+		cfg.Influx.Password = dc.Influx.Password
+	}
+
+	if cfg.SinkKafkaStream == "" {
+		cfg.SinkKafkaStream = dc.SinkKafkaStream
+	}
+
+	if cfg.KubeNamespace == "" {
+		cfg.KubeNamespace = dc.KubeNamespace
+	}
+
+	return &cfg, nil
+}

modules/flink/driver.go

@@ -0,0 +1,37 @@
+package flink
+
+import (
+	"encoding/json"
+
+	"github.com/goto/entropy/core/module"
+	"github.com/goto/entropy/modules/kubernetes"
+	"github.com/goto/entropy/pkg/errors"
+)
+
+type flinkDriver struct {
+	conf driverConf
+}
+
+type driverConf struct {
+	Influx          Influx `json:"influx,omitempty"`
+	SinkKafkaStream string `json:"sink_kafka_stream,omitempty"`
+	KubeNamespace   string `json:"kube_namespace,omitempty"`
+}
+
+type Output struct {
+	KubeCluster     kubernetes.Output `json:"kube_cluster,omitempty"`
+	KubeNamespace   string            `json:"kube_namespace,omitempty"`
+	Influx          Influx            `json:"influx,omitempty"`
+	SinkKafkaStream string            `json:"sink_kafka_stream,omitempty"`
+}
+
+func readOutputData(exr module.ExpandedResource) (*Output, error) {
+	var curOut Output
+	if len(exr.Resource.State.Output) == 0 {
+		return &curOut, nil
+	}
+	if err := json.Unmarshal(exr.Resource.State.Output, &curOut); err != nil {
+		return nil, errors.ErrInternal.WithMsgf("corrupted output").WithCausef(err.Error())
+	}
+	return &curOut, nil
+}

modules/flink/driver_output.go

@@ -0,0 +1,38 @@
+package flink
+
+import (
+	"context"
+	"encoding/json"
+
+	"github.com/goto/entropy/core/module"
+	"github.com/goto/entropy/modules"
+	"github.com/goto/entropy/modules/kubernetes"
+	"github.com/goto/entropy/pkg/errors"
+)
+
+func (fd *flinkDriver) Output(ctx context.Context, exr module.ExpandedResource) (json.RawMessage, error) {
+	output, err := readOutputData(exr)
+	if err != nil {
+		return nil, err
+	}
+
+	conf, err := readConfig(exr.Resource, exr.Resource.Spec.Configs, fd.conf)
+	if err != nil {
+		if errors.Is(err, errors.ErrInvalid) {
+			return nil, err
+		}
+		return nil, errors.ErrInternal.WithCausef(err.Error())
+	}
+
+	var kubeOut kubernetes.Output
+	if err := json.Unmarshal(exr.Dependencies[keyKubeDependency].Output, &kubeOut); err != nil {
+		return nil, errors.ErrInternal.WithMsgf("invalid kube state").WithCausef(err.Error())
+	}
+
+	output.KubeCluster = kubeOut
+	output.Influx = conf.Influx
+	output.KubeNamespace = conf.KubeNamespace
+	output.SinkKafkaStream = conf.SinkKafkaStream
+
+	return modules.MustJSON(output), nil
+}

modules/flink/driver_plan.go

@@ -0,0 +1,27 @@
+package flink
+
+import (
+	"context"
+
+	"github.com/goto/entropy/core/module"
+	"github.com/goto/entropy/core/resource"
+)
+
+func (fd *flinkDriver) Plan(ctx context.Context, res module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) {
+	res.Resource.Spec = resource.Spec{
+		Configs:      act.Params,
+		Dependencies: res.Spec.Dependencies,
+	}
+
+	output, err := fd.Output(ctx, res)
+	if err != nil {
+		return nil, err
+	}
+
+	res.Resource.State = resource.State{
+		Status: resource.StatusCompleted,
+		Output: output,
+	}
+
+	return &res.Resource, nil
+}

modules/flink/driver_sync.go

@@ -0,0 +1,16 @@
+package flink
+
+import (
+	"context"
+
+	"github.com/goto/entropy/core/module"
+	"github.com/goto/entropy/core/resource"
+)
+
+func (*flinkDriver) Sync(_ context.Context, res module.ExpandedResource) (*resource.State, error) {
+	return &resource.State{
+		Status:     resource.StatusCompleted,
+		Output:     res.Resource.State.Output,
+		ModuleData: nil,
+	}, nil
+}

modules/flink/module.go

@@ -0,0 +1,33 @@
+package flink
+
+import (
+	_ "embed"
+	"encoding/json"
+
+	"github.com/goto/entropy/core/module"
+	"github.com/goto/entropy/pkg/errors"
+)
+
+const (
+	keyKubeDependency = "kube_cluster"
+)
+
+var Module = module.Descriptor{
+	Kind: "flink",
+	Actions: []module.ActionDesc{
+		{
+			Name: module.CreateAction,
+		},
+		{
+			Name: module.UpdateAction,
+		},
+	},
+	DriverFactory: func(conf json.RawMessage) (module.Driver, error) {
+		fd := &flinkDriver{}
+		err := json.Unmarshal(conf, &fd)
+		if err != nil {
+			return nil, errors.ErrInvalid.WithMsgf("failed to unmarshal module config: %v", err)
+		}
+		return fd, nil
+	},
+}

modules/flink/schema/config.json

@@ -0,0 +1,27 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "$id": "http://json-schema.org/draft-07/schema#",
+  "type": "object",
+  "properties": {
+    "kube_namespace": {
+      "type": "string"
+    },
+    "influx": {
+      "type": "object",
+      "properties": {
+        "url": {
+          "type": "string"
+        },
+        "username": {
+          "type": "string"
+        },
+        "password": {
+          "type": "string"
+        }
+      }
+    },
+    "sink_kafka_stream_name": {
+      "type": "string"
+    }
+  }
+}

test/e2e_test/firehose_test.go

@@ -36,7 +36,7 @@ func (s *FirehoseTestSuite) SetupTest() {
 
 	modules, err := s.moduleClient.ListModules(s.ctx, &entropyv1beta1.ListModulesRequest{})
 	s.Require().NoError(err)
-	s.Require().Equal(6, len(modules.GetModules()))
+	s.Require().Equal(9, len(modules.GetModules()))
 
 	resources, err := s.resourceClient.ListResources(s.ctx, &entropyv1beta1.ListResourcesRequest{
 		Kind: "kubernetes",