Skip to content

Commit 873e9e3

Browse files
raulbraulb
authored
feat: add processor describe (#2089)
* feat: add processor describe * check empty cases
1 parent d8169e0 commit 873e9e3

File tree

5 files changed

+186
-1
lines changed

5 files changed

+186
-1
lines changed

Diff for: ‎cmd/conduit/internal/print_utils.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,7 @@ func DisplayProcessors(processors []*apiv1.Processor, indent int) {
6363
fmt.Printf("%s- ID: %s\n", Indentation(indent+1), p.Id)
6464
fmt.Printf("%sPlugin: %s\n", Indentation(indent+2), p.Plugin)
6565

66-
if p.Condition != "" {
66+
if !IsEmpty(p.Condition) {
6767
fmt.Printf("%sCondition: %s\n", Indentation(indent+2), p.Condition)
6868
}
6969

Diff for: ‎cmd/conduit/root/processors/describe.go

+113
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,113 @@
1+
// Copyright © 2025 Meroxa, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package processors
16+
17+
import (
18+
"context"
19+
"fmt"
20+
21+
"github.com/conduitio/conduit/cmd/conduit/api"
22+
"github.com/conduitio/conduit/cmd/conduit/cecdysis"
23+
"github.com/conduitio/conduit/cmd/conduit/internal"
24+
"github.com/conduitio/conduit/pkg/foundation/cerrors"
25+
apiv1 "github.com/conduitio/conduit/proto/api/v1"
26+
"github.com/conduitio/ecdysis"
27+
)
28+
29+
var (
30+
_ cecdysis.CommandWithExecuteWithClient = (*DescribeCommand)(nil)
31+
_ ecdysis.CommandWithAliases = (*DescribeCommand)(nil)
32+
_ ecdysis.CommandWithDocs = (*DescribeCommand)(nil)
33+
_ ecdysis.CommandWithArgs = (*DescribeCommand)(nil)
34+
)
35+
36+
type DescribeArgs struct {
37+
ProcessorID string
38+
}
39+
40+
type DescribeCommand struct {
41+
args DescribeArgs
42+
}
43+
44+
func (c *DescribeCommand) Usage() string { return "describe" }
45+
46+
func (c *DescribeCommand) Docs() ecdysis.Docs {
47+
return ecdysis.Docs{
48+
Short: "Describe an existing processor",
49+
Long: `This command requires Conduit to be already running since it will describe a processor registered
50+
by Conduit. You can list existing processors with the 'conduit processors list' command.`,
51+
Example: "conduit processors describe pipeline-processor\n" +
52+
"conduit processor desc connector-processor",
53+
}
54+
}
55+
56+
func (c *DescribeCommand) Aliases() []string { return []string{"desc"} }
57+
58+
func (c *DescribeCommand) Args(args []string) error {
59+
if len(args) == 0 {
60+
return cerrors.Errorf("requires a processor ID")
61+
}
62+
63+
if len(args) > 1 {
64+
return cerrors.Errorf("too many arguments")
65+
}
66+
67+
c.args.ProcessorID = args[0]
68+
return nil
69+
}
70+
71+
func (c *DescribeCommand) ExecuteWithClient(ctx context.Context, client *api.Client) error {
72+
resp, err := client.ProcessorServiceClient.GetProcessor(ctx, &apiv1.GetProcessorRequest{
73+
Id: c.args.ProcessorID,
74+
})
75+
if err != nil {
76+
return fmt.Errorf("failed to get processor: %w", err)
77+
}
78+
79+
displayProcessor(resp.Processor)
80+
return nil
81+
}
82+
83+
func displayProcessor(p *apiv1.Processor) {
84+
fmt.Printf("ID: %s\n", p.Id)
85+
fmt.Printf("Plugin: %s\n", p.Plugin)
86+
87+
processorType := "Processor "
88+
switch p.Parent.Type.String() {
89+
case "TYPE_PIPELINE":
90+
processorType += "for Pipeline"
91+
case "TYPE_CONNECTOR":
92+
processorType += "for Connector"
93+
default:
94+
processorType += "associated to"
95+
}
96+
97+
fmt.Printf("%s: %s\n", processorType, p.Parent.Id)
98+
99+
if !internal.IsEmpty(p.Condition) {
100+
fmt.Printf("Condition: %s\n", p.Condition)
101+
}
102+
103+
if len(p.Config.Settings) > 0 {
104+
fmt.Println("Config:")
105+
for name, value := range p.Config.Settings {
106+
fmt.Printf("%s%s: %s\n", internal.Indentation(1), name, value)
107+
}
108+
}
109+
fmt.Printf("Workers: %d\n", p.Config.Workers)
110+
111+
fmt.Printf("Created At: %s\n", internal.PrintTime(p.CreatedAt))
112+
fmt.Printf("Updated At: %s\n", internal.PrintTime(p.UpdatedAt))
113+
}

Diff for: ‎cmd/conduit/root/processors/describe_test.go

+56
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
// Copyright © 2025 Meroxa, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package processors
16+
17+
import (
18+
"testing"
19+
20+
"github.com/matryer/is"
21+
)
22+
23+
func TestDescribeExecutionNoArgs(t *testing.T) {
24+
is := is.New(t)
25+
26+
c := DescribeCommand{}
27+
err := c.Args([]string{})
28+
29+
expected := "requires a processor ID"
30+
31+
is.True(err != nil)
32+
is.Equal(err.Error(), expected)
33+
}
34+
35+
func TestDescribeExecutionMultipleArgs(t *testing.T) {
36+
is := is.New(t)
37+
38+
c := DescribeCommand{}
39+
err := c.Args([]string{"foo", "bar"})
40+
41+
expected := "too many arguments"
42+
43+
is.True(err != nil)
44+
is.Equal(err.Error(), expected)
45+
}
46+
47+
func TestDescribeExecutionCorrectArgs(t *testing.T) {
48+
is := is.New(t)
49+
processorID := "my-processor"
50+
51+
c := DescribeCommand{}
52+
err := c.Args([]string{processorID})
53+
54+
is.NoErr(err)
55+
is.Equal(c.args.ProcessorID, processorID)
56+
}

Diff for: ‎cmd/conduit/root/processors/list.go

+15
Original file line numberDiff line numberDiff line change
@@ -75,16 +75,31 @@ func displayProcessors(processors []*apiv1.Processor) {
7575
{Align: simpletable.AlignCenter, Text: "ID"},
7676
{Align: simpletable.AlignCenter, Text: "PLUGIN"},
7777
{Align: simpletable.AlignCenter, Text: "CONDITION"},
78+
{Align: simpletable.AlignCenter, Text: "TYPE"},
7879
{Align: simpletable.AlignCenter, Text: "CREATED"},
7980
{Align: simpletable.AlignCenter, Text: "LAST_UPDATED"},
8081
},
8182
}
8283

8384
for _, p := range processors {
85+
var processorType string
86+
87+
switch p.Parent.Type.String() {
88+
case "TYPE_PIPELINE":
89+
processorType = "Pipeline"
90+
case "TYPE_CONNECTOR":
91+
processorType = "Connector"
92+
default:
93+
processorType = "Unknown"
94+
}
95+
96+
processorType = fmt.Sprintf("%s (%s)", processorType, p.Parent.Id)
97+
8498
r := []*simpletable.Cell{
8599
{Align: simpletable.AlignLeft, Text: p.Id},
86100
{Align: simpletable.AlignLeft, Text: p.Plugin},
87101
{Align: simpletable.AlignLeft, Text: p.Condition},
102+
{Align: simpletable.AlignLeft, Text: processorType},
88103
{Align: simpletable.AlignLeft, Text: internal.PrintTime(p.CreatedAt)},
89104
{Align: simpletable.AlignLeft, Text: internal.PrintTime(p.UpdatedAt)},
90105
}

Diff for: ‎cmd/conduit/root/processors/processors.go

+1
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ func (c *ProcessorsCommand) Aliases() []string { return []string{"processor"} }
3131
func (c *ProcessorsCommand) SubCommands() []ecdysis.Command {
3232
return []ecdysis.Command{
3333
&ListCommand{},
34+
&DescribeCommand{},
3435
}
3536
}
3637

0 commit comments

Comments
 (0)