Basic conduit init and conduit pipelines init commands

cmd/cli/cli.go

@@ -0,0 +1,147 @@
+// Copyright © 2024 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cli
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/conduitio/conduit/pkg/conduit"
+	"github.com/spf13/cobra"
+)
+
+var (
+	initArgs          InitArgs
+	pipelinesInitArgs PipelinesInitArgs
+)
+
+type Instance struct {
+	rootCmd *cobra.Command
+}
+
+// New creates a new CLI Instance.
+func New() *Instance {
+	return &Instance{
+		rootCmd: buildRootCmd(),
+	}
+}
+
+func (i *Instance) Run() {
+	if err := i.rootCmd.Execute(); err != nil {
+		_, _ = fmt.Fprintf(os.Stderr, "%v\n", err)
+		os.Exit(1)
+	}
+}
+
+func buildRootCmd() *cobra.Command {
+	cfg := conduit.DefaultConfig()
+
+	cmd := &cobra.Command{
+		Use:     "conduit",
+		Short:   "Conduit CLI",
+		Long:    "Conduit CLI is a command-line that helps you interact with and manage Conduit.",
+		Version: conduit.Version(true),
+		Run: func(cmd *cobra.Command, args []string) {
+			e := &conduit.Entrypoint{}
+			e.Serve(cfg)
+		},
+	}
+	cmd.CompletionOptions.DisableDefaultCmd = true
+	conduit.Flags(&cfg).VisitAll(cmd.Flags().AddGoFlag)
+
+	// init
+	cmd.AddCommand(buildInitCmd())
+
+	// pipelines
+	cmd.AddGroup(&cobra.Group{
+		ID:    "pipelines",
+		Title: "Pipelines",
+	})
+	cmd.AddCommand(buildPipelinesCmd())
+
+	return cmd
+}
+
+func buildInitCmd() *cobra.Command {
+	initCmd := &cobra.Command{
+		Use:   "init",
+		Short: "Initialize Conduit with a configuration file and directories.",
+		Args:  cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			return NewConduitInit(initArgs).Run()
+		},
+	}
+	initCmd.Flags().StringVar(
+		&initArgs.Path,
+		"config.path",
+		"",
+		"path where Conduit will be initialized",
+	)
+
+	return initCmd
+}
+
+func buildPipelinesCmd() *cobra.Command {
+	pipelinesCmd := &cobra.Command{
+		Use:     "pipelines",
+		Short:   "Initialize and manage pipelines",
+		Args:    cobra.NoArgs,
+		GroupID: "pipelines",
+	}
+
+	pipelinesCmd.AddCommand(buildPipelinesInitCmd())
+
+	return pipelinesCmd
+}
+
+func buildPipelinesInitCmd() *cobra.Command {
+	pipelinesInitCmd := &cobra.Command{
+		Use:   "init [pipeline-name]",
+		Short: "Initialize an example pipeline.",
+		Long: `Initialize a pipeline configuration file, with all of parameters for source and destination connectors 
+initialized and described. The source and destination connector can be chosen via flags. If no connectors are chosen, then
+a simple and runnable generator-to-log pipeline is configured.`,
+		Args:    cobra.MaximumNArgs(1),
+		Example: "  conduit pipelines init awesome-pipeline-name --source postgres --destination kafka --path pipelines/pg-to-kafka.yaml",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			if len(args) > 0 {
+				pipelinesInitArgs.Name = args[0]
+			}
+			return NewPipelinesInit(pipelinesInitArgs).Run()
+		},
+	}
+
+	// Add flags to pipelines init command
+	pipelinesInitCmd.Flags().StringVar(
+		&pipelinesInitArgs.Source,
+		"source",
+		"",
+		"Source connector (any of the built-in connectors).",
+	)
+	pipelinesInitCmd.Flags().StringVar(
+		&pipelinesInitArgs.Destination,
+		"destination",
+		"",
+		"Destination connector (any of the built-in connectors).",
+	)
+	pipelinesInitCmd.Flags().StringVar(
+		&pipelinesInitArgs.Path,
+		"pipelines.path",
+		"./pipelines",
+		"Path where the pipeline will be saved.",
+	)
+
+	return pipelinesInitCmd
+}

cmd/cli/conduit_init.go

@@ -0,0 +1,129 @@
+// Copyright © 2024 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cli
+
+import (
+	"flag"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/conduitio/conduit/cmd/cli/internal"
+	"github.com/conduitio/conduit/pkg/conduit"
+	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/yaml/v3"
+)
+
+type InitArgs struct {
+	Path string
+}
+
+type ConduitInit struct {
+	args InitArgs
+}
+
+func NewConduitInit(args InitArgs) *ConduitInit {
+	return &ConduitInit{args: args}
+}
+
+func (i *ConduitInit) Run() error {
+	err := i.createDirs()
+	if err != nil {
+		return err
+	}
+
+	err = i.createConfigYAML()
+	if err != nil {
+		return fmt.Errorf("failed to create config YAML: %w", err)
+	}
+
+	fmt.Println(`
+Conduit has been initialized!
+
+To quickly create an example pipeline, run 'conduit pipelines init'.
+To see how you can customize your first pipeline, run 'conduit pipelines init --help'.`)
+
+	return nil
+}
+
+func (i *ConduitInit) createConfigYAML() error {
+	cfgYAML := internal.NewYAMLTree()
+	i.conduitCfgFlags().VisitAll(func(f *flag.Flag) {
+		if i.isHiddenFlag(f.Name) {
+			return // hide flag from output
+		}
+		cfgYAML.Insert(f.Name, f.DefValue, f.Usage)
+	})
+
+	yamlData, err := yaml.Marshal(cfgYAML.Root)
+	if err != nil {
+		return cerrors.Errorf("error marshaling YAML: %w\n", err)
+	}
+
+	path := filepath.Join(i.path(), "conduit.yaml")
+	err = os.WriteFile(path, yamlData, 0o600)
+	if err != nil {
+		return cerrors.Errorf("error writing conduit.yaml: %w", err)
+	}
+	fmt.Printf("Configuration file written to %v\n", path)
+
+	return nil
+}
+
+func (i *ConduitInit) createDirs() error {
+	dirs := []string{"processors", "connectors", "pipelines"}
+
+	for _, dir := range dirs {
+		path := filepath.Join(i.path(), dir)
+
+		// Attempt to create the directory, skipping if it already exists
+		if err := os.Mkdir(path, os.ModePerm); err != nil {
+			if os.IsExist(err) {
+				fmt.Printf("Directory '%s' already exists, skipping...\n", path)
+				continue
+			}
+			return fmt.Errorf("failed to create directory '%s': %w", path, err)
+		}
+
+		fmt.Printf("Created directory: %s\n", path)
+	}
+
+	return nil
+}
+
+func (i *ConduitInit) isHiddenFlag(name string) bool {
+	return name == "dev" ||
+		strings.HasPrefix(name, "dev.") ||
+		conduit.DeprecatedFlags[name]
+}
+
+func (i *ConduitInit) conduitCfgFlags() *flag.FlagSet {
+	cfg := conduit.DefaultConfigWithBasePath(i.path())
+	return conduit.Flags(&cfg)
+}
+
+func (i *ConduitInit) path() string {
+	if i.args.Path != "" {
+		return i.args.Path
+	}
+
+	path, err := os.Getwd()
+	if err != nil {
+		panic(cerrors.Errorf("failed to get current working directory: %w", err))
+	}
+
+	return path
+}

cmd/cli/internal/yaml.go

@@ -0,0 +1,85 @@
+// Copyright © 2024 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package internal
+
+import (
+	"strings"
+
+	"github.com/conduitio/yaml/v3"
+)
+
+// YAMLTree represents a YAML document.
+// It makes it possible to insert value nodes with comments.
+type YAMLTree struct {
+	Root *yaml.Node
+}
+
+func NewYAMLTree() *YAMLTree {
+	return &YAMLTree{
+		Root: &yaml.Node{
+			Kind: yaml.MappingNode,
+		},
+	}
+}
+
+// Insert adds a path with a value to the tree
+func (t *YAMLTree) Insert(path, value, comment string) {
+	parts := strings.Split(path, ".")
+	current := t.Root
+
+	// For each part of the path
+	for i, part := range parts {
+		// Create key node
+		keyNode := &yaml.Node{
+			Kind:  yaml.ScalarNode,
+			Value: part,
+		}
+
+		// Find or create value node
+		var valueNode *yaml.Node
+		found := false
+
+		// Look for existing key in current mapping
+		for i := 0; i < len(current.Content); i += 2 {
+			if current.Content[i].Value == part {
+				valueNode = current.Content[i+1]
+				found = true
+				break
+			}
+		}
+
+		// If not found, create new node
+		if !found {
+			// If this is the last part, create scalar value node
+			if i == len(parts)-1 {
+				valueNode = &yaml.Node{
+					Kind:  yaml.ScalarNode,
+					Value: value,
+				}
+				keyNode.HeadComment = comment
+			} else {
+				// Otherwise create mapping node for nesting
+				valueNode = &yaml.Node{
+					Kind: yaml.MappingNode,
+				}
+			}
+			// Add key-value pair to current node's content
+			current.Content = append(current.Content, keyNode, valueNode)
+		}
+
+		// Move to next level
+		current = valueNode
+	}
+}

cmd/cli/pipeline.tmpl

@@ -0,0 +1,30 @@
+version: "2.2"
+pipelines:
+  - id: example-pipeline
+    status: running
+    name: "{{ .Name }}"
+    connectors:
+      - id: example-source
+        type: source
+        plugin: "{{ .SourceSpec.Name }}"
+        {{ if gt (len .SourceSpec.Params) 0 -}}
+        settings:
+          {{- range $name, $param := .SourceSpec.Params }}
+          {{ formatParameterDescriptionYAML $param.Description }}
+          # Type: {{ $param.Type }}
+          # {{ formatParameterRequired $param }}
+          {{ $name }}: {{ formatParameterValueYAML $param.Default }}
+          {{- end }}
+        {{- end }}
+      - id: example-destination
+        type: destination
+        plugin: "{{ .DestinationSpec.Name }}"
+        {{ if gt (len .DestinationSpec.Params) 0 -}}
+        settings:
+          {{- range $name, $param := .DestinationSpec.Params }}
+          {{ formatParameterDescriptionYAML $param.Description }}
+          # Type: {{ $param.Type }}
+          # {{ formatParameterRequired $param }}
+          {{ $name }}: {{ formatParameterValueYAML $param.Default }}
+          {{- end }}
+        {{- end }}