feat(turbine): logically nest functions under applications (#267)

* feat(turbine): logically nest functions under applications

cmd/meroxa/root/apps/list_test.go

@@ -24,6 +24,8 @@ import (
 	"testing"
 
 	"github.com/golang/mock/gomock"
+	"github.com/volatiletech/null/v8"
+
 	"github.com/meroxa/cli/log"
 	"github.com/meroxa/cli/utils"
 	"github.com/meroxa/meroxa-go/pkg/meroxa"
@@ -42,8 +44,8 @@ func TestListAppsExecution(t *testing.T) {
 		Language: GoLang,
 		Status:   meroxa.ApplicationStatus{State: meroxa.ApplicationStateReady},
 		Functions: []meroxa.FunctionIdentifier{
-			{Name: "one"},
-			{Name: "two"},
+			{Name: null.StringFrom("one")},
+			{Name: null.StringFrom("two")},
 		},
 	}
 

cmd/meroxa/root/functions/create.go

@@ -6,6 +6,8 @@ import (
 	"strings"
 
 	"github.com/mattn/go-shellwords"
+	"github.com/volatiletech/null/v8"
+
 	"github.com/meroxa/cli/cmd/meroxa/builder"
 	"github.com/meroxa/cli/log"
 	"github.com/meroxa/meroxa-go/pkg/meroxa"
@@ -39,6 +41,7 @@ type Create struct {
 		Args        string   `long:"args" usage:"Arguments to the entrypoint"`
 		EnvVars     []string `long:"env" usage:"List of environment variables to set in the function"`
 		Pipeline    string   `long:"pipeline" usage:"pipeline name to attach function to" required:"true"`
+		Application string   `long:"app" usage:"application name or UUID to which this function belongs" required:"true"`
 	}
 }
 
@@ -49,10 +52,10 @@ func (c *Create) Usage() string {
 func (c *Create) Docs() builder.Docs {
 	return builder.Docs{
 		Short: "Create a function",
-		Long:  "Use `functions create` to create a function to process records from an input steram (--input-stream)",
+		Long:  "Use `functions create` to create a function to process records from an input stream (--input-stream)",
 		Example: `
-meroxa functions create [NAME] --input-stream connector-output-stream --image myimage --pipeline my-pipeline
-meroxa functions create [NAME] --input-stream connector-output-stream --image myimage --pipeline my-pipeline --env FOO=BAR --env BAR=BAZ
+meroxa functions create [NAME] --input-stream connector-output-stream --image myimage --app my-app
+meroxa functions create [NAME] --input-stream connector-output-stream --image myimage --app my-app --env FOO=BAR --env BAR=BAZ
 `,
 	}
 }
@@ -86,7 +89,10 @@ func (c *Create) Execute(ctx context.Context) error {
 			Name:        c.args.Name,
 			InputStream: c.flags.InputStream,
 			Pipeline: meroxa.PipelineIdentifier{
-				Name: c.flags.Pipeline,
+				Name: null.StringFrom(c.flags.Pipeline),
+			},
+			Application: meroxa.ApplicationIdentifier{
+				Name: null.StringFrom(c.flags.Application),
 			},
 			Image:   c.flags.Image,
 			Command: command,

cmd/meroxa/root/functions/describe.go

@@ -19,7 +19,7 @@ var (
 )
 
 type describeFunctionClient interface {
-	GetFunction(ctx context.Context, nameOrUUID string) (*meroxa.Function, error)
+	GetFunction(ctx context.Context, appNameOrUUID, nameOrUUID string) (*meroxa.Function, error)
 }
 
 type Describe struct {
@@ -29,6 +29,10 @@ type Describe struct {
 	args struct {
 		NameOrUUID string
 	}
+
+	flags struct {
+		Application string `long:"app" usage:"application name or UUID to which this function belongs" required:"true"`
+	}
 }
 
 func (d *Describe) Usage() string {
@@ -42,7 +46,7 @@ func (d *Describe) Docs() builder.Docs {
 }
 
 func (d *Describe) Execute(ctx context.Context) error {
-	fun, err := d.client.GetFunction(ctx, d.args.NameOrUUID)
+	fun, err := d.client.GetFunction(ctx, d.flags.Application, d.args.NameOrUUID)
 	if err != nil {
 		return err
 	}

cmd/meroxa/root/functions/functions.go

@@ -14,7 +14,7 @@ var (
 	_ builder.CommandWithDocs        = (*Functions)(nil)
 	_ builder.CommandWithFeatureFlag = (*Functions)(nil)
 	_ builder.CommandWithSubCommands = (*Functions)(nil)
-	_ builder.CommandWithHidden      = (*Functions)(nil)
+	_ builder.CommandWithHidden      = (*Functions)(nil) // for internal use only, will always be hidden
 )
 
 func (*Functions) Usage() string {
@@ -26,7 +26,7 @@ func (*Functions) Hidden() bool {
 }
 
 func (*Functions) FeatureFlag() (string, error) {
-	return "functions", fmt.Errorf(`no access to the Meroxa functions feature`)
+	return "turbine", fmt.Errorf(`no access to the Meroxa Data applications feature`)
 }
 
 func (*Functions) Docs() builder.Docs {

cmd/meroxa/root/functions/list.go

@@ -19,19 +19,40 @@ var (
 )
 
 type listFunctionClient interface {
-	ListFunctions(ctx context.Context) ([]*meroxa.Function, error)
+	ListApplications(ctx context.Context) ([]*meroxa.Application, error)
+	ListFunctions(ctx context.Context, appNameOrUUID string) ([]*meroxa.Function, error)
 }
 
 type List struct {
 	client      listFunctionClient
 	logger      log.Logger
 	hideHeaders bool
+
+	flags struct {
+		Application string `long:"app" usage:"application name or UUID to which this function belongs"`
+	}
 }
 
 func (l *List) Execute(ctx context.Context) error {
-	funs, err := l.client.ListFunctions(ctx)
-	if err != nil {
-		return err
+	var err error
+	funs := make([]*meroxa.Function, 0)
+	if l.flags.Application != "" {
+		funs, err = l.client.ListFunctions(ctx, l.flags.Application)
+		if err != nil {
+			return err
+		}
+	} else {
+		apps, err := l.client.ListApplications(ctx)
+		if err != nil {
+			return err
+		}
+		for _, app := range apps {
+			fs, err := l.client.ListFunctions(ctx, app.UUID)
+			if err != nil {
+				return err
+			}
+			funs = append(funs, fs...)
+		}
 	}
 
 	l.logger.JSON(ctx, funs)

cmd/meroxa/root/functions/logs.go

@@ -20,7 +20,7 @@ var (
 )
 
 type functionLogsClient interface {
-	GetFunctionLogs(ctx context.Context, nameOrUUID string) (*http.Response, error)
+	GetFunctionLogs(ctx context.Context, appNameOrUUID, nameOrUUID string) (*http.Response, error)
 }
 
 type Logs struct {
@@ -30,6 +30,10 @@ type Logs struct {
 	args struct {
 		NameOrUUID string
 	}
+
+	flags struct {
+		Application string `long:"app" usage:"application name or UUID to which this function belongs" required:"true"`
+	}
 }
 
 func (l *Logs) Usage() string {
@@ -43,7 +47,7 @@ func (l *Logs) Docs() builder.Docs {
 }
 
 func (l *Logs) Execute(ctx context.Context) error {
-	resp, err := l.client.GetFunctionLogs(ctx, l.args.NameOrUUID)
+	resp, err := l.client.GetFunctionLogs(ctx, l.flags.Application, l.args.NameOrUUID)
 
 	if err != nil {
 		return err

cmd/meroxa/root/functions/remove.go

@@ -20,7 +20,7 @@ var (
 )
 
 type removeFunctionClient interface {
-	DeleteFunction(ctx context.Context, nameOrUUID string) (*meroxa.Function, error)
+	DeleteFunction(ctx context.Context, appNameOrUUID, nameOrUUID string) (*meroxa.Function, error)
 }
 
 type Remove struct {
@@ -30,6 +30,10 @@ type Remove struct {
 	args struct {
 		NameOrUUID string
 	}
+
+	flags struct {
+		Application string `long:"app" usage:"application name or UUID to which this function belongs" required:"true"`
+	}
 }
 
 func (r *Remove) Usage() string {
@@ -49,7 +53,7 @@ func (r *Remove) ValueToConfirm(_ context.Context) (wantInput string) {
 func (r *Remove) Execute(ctx context.Context) error {
 	r.logger.Infof(ctx, "Function %q is being removed...", r.args.NameOrUUID)
 
-	e, err := r.client.DeleteFunction(ctx, r.args.NameOrUUID)
+	e, err := r.client.DeleteFunction(ctx, r.flags.Application, r.args.NameOrUUID)
 	if err != nil {
 		return err
 	}

go.mod

@@ -43,7 +43,9 @@ require (
 	github.com/docker/distribution v2.8.0+incompatible // indirect
 	github.com/docker/go-connections v0.4.0 // indirect
 	github.com/docker/go-units v0.4.0 // indirect
+	github.com/friendsofgo/errors v0.9.2 // indirect
 	github.com/fsnotify/fsnotify v1.4.9 // indirect
+	github.com/gofrs/uuid v4.2.0+incompatible // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
@@ -69,6 +71,10 @@ require (
 	github.com/spf13/cast v1.3.1 // indirect
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
 	github.com/subosito/gotenv v1.2.0 // indirect
+	github.com/volatiletech/inflect v0.0.1 // indirect
+	github.com/volatiletech/null/v8 v8.1.2 // indirect
+	github.com/volatiletech/randomize v0.0.1 // indirect
+	github.com/volatiletech/strmangle v0.0.2 // indirect
 	go.opencensus.io v0.23.0 // indirect
 	golang.org/x/net v0.0.0-20220225172249-27dd8689420f // indirect
 	golang.org/x/sys v0.0.0-20220227234510-4e6760a101f9 // indirect
@@ -82,3 +88,5 @@ require (
 	gopkg.in/ini.v1 v1.62.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 )
+
+replace github.com/meroxa/meroxa-go => ../meroxa-go

go.sum

@@ -318,6 +318,8 @@ github.com/fatih/color v1.9.0 h1:8xPHl4/q1VyqGIPif1F+1V3Y3lSmrq01EabUW3CoW5s=
 github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL+zU=
 github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
 github.com/frankban/quicktest v1.11.3/go.mod h1:wRf/ReqHper53s+kmmSZizM8NamnL3IM0I9ntUbOk+k=
+github.com/friendsofgo/errors v0.9.2 h1:X6NYxef4efCBdwI7BgS820zFaN7Cphrmb+Pljdzjtgk=
+github.com/friendsofgo/errors v0.9.2/go.mod h1:yCvFW5AkDIL9qn7suHVLiI/gH228n7PC4Pn44IGoTOI=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
@@ -356,6 +358,9 @@ github.com/godbus/dbus v0.0.0-20190422162347-ade71ed3457e/go.mod h1:bBOAhwG1umN6
 github.com/godbus/dbus/v5 v5.0.3/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/godbus/dbus/v5 v5.0.6/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
+github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
+github.com/gofrs/uuid v4.2.0+incompatible h1:yyYWMnhkhrKwwr8gAOcOCYxOOscHgDS9yZgBrnJfGa0=
+github.com/gofrs/uuid v4.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
 github.com/gogo/googleapis v1.2.0/go.mod h1:Njal3psf3qN6dwBtQfUmBZh2ybovJ0tlu3o/AC7HYjU=
 github.com/gogo/googleapis v1.4.0/go.mod h1:5YRNX2z1oM5gXdAkurHa942MDgEJyk02w4OecKY87+c=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
@@ -569,9 +574,6 @@ github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
 github.com/maxbrunsfeld/counterfeiter/v6 v6.2.2/go.mod h1:eD9eIE7cdwcMi9rYluz88Jz2VyhSmden33/aXg4oVIY=
 github.com/meroxa/funtime v0.0.0-20220113012133-85e6e898fc73/go.mod h1:K2y2GvcA4Cg3dJtckcwYWnwnJzF63FDdtAQI0fToU0Q=
-github.com/meroxa/meroxa-go v0.0.0-20220208195203-71ddc3133fab/go.mod h1:HDFszURCM1cOpKE699o5Hs0T2tEIXqY+vFcsur3RiwY=
-github.com/meroxa/meroxa-go v0.0.0-20220302153558-e3b3dc31559c h1:Xgq1zRKeonrGZc0NATJVDYyXniuZJftpJ2dOi6SpHB4=
-github.com/meroxa/meroxa-go v0.0.0-20220302153558-e3b3dc31559c/go.mod h1:ab2rHsqdQ25tYbIohjRGNujKHu6yGhV5ALAi4GE0Ehw=
 github.com/meroxa/turbine v0.0.0-20220301211444-5662995ad65f h1:fWCA+cO4aPIzePOuz4drdOQAgUTM6UG7lRYgtbWLHv0=
 github.com/meroxa/turbine v0.0.0-20220301211444-5662995ad65f/go.mod h1:4A2E9icHDi3x4Flp9CunWyRBvNkZy6h8MS8zLvWUlmw=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
@@ -805,6 +807,15 @@ github.com/vishvananda/netlink v1.1.1-0.20201029203352-d40f9887b852/go.mod h1:tw
 github.com/vishvananda/netns v0.0.0-20180720170159-13995c7128cc/go.mod h1:ZjcWmFBXmLKZu9Nxj3WKYEafiSqer2rnvPr0en9UNpI=
 github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17pCcGlemwknint6hfoeCVQrEMVwxRLRjXpq+BU=
 github.com/vishvananda/netns v0.0.0-20200728191858-db3c7e526aae/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0=
+github.com/volatiletech/inflect v0.0.1 h1:2a6FcMQyhmPZcLa+uet3VJ8gLn/9svWhJxJYwvE8KsU=
+github.com/volatiletech/inflect v0.0.1/go.mod h1:IBti31tG6phkHitLlr5j7shC5SOo//x0AjDzaJU1PLA=
+github.com/volatiletech/null/v8 v8.1.2 h1:kiTiX1PpwvuugKwfvUNX/SU/5A2KGZMXfGD0DUHdKEI=
+github.com/volatiletech/null/v8 v8.1.2/go.mod h1:98DbwNoKEpRrYtGjWFctievIfm4n4MxG0A6EBUcoS5g=
+github.com/volatiletech/randomize v0.0.1 h1:eE5yajattWqTB2/eN8df4dw+8jwAzBtbdo5sbWC4nMk=
+github.com/volatiletech/randomize v0.0.1/go.mod h1:GN3U0QYqfZ9FOJ67bzax1cqZ5q2xuj2mXrXBjWaRTlY=
+github.com/volatiletech/strmangle v0.0.1/go.mod h1:F6RA6IkB5vq0yTG4GQ0UsbbRcl3ni9P76i+JrTBKFFg=
+github.com/volatiletech/strmangle v0.0.2 h1:amhpV9ATyq1DtkQ2D8WF94uqGpkYYmSmx7X2QIner/A=
+github.com/volatiletech/strmangle v0.0.2/go.mod h1:F6RA6IkB5vq0yTG4GQ0UsbbRcl3ni9P76i+JrTBKFFg=
 github.com/willf/bitset v1.1.11-0.20200630133818-d5bec3311243/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4=
 github.com/willf/bitset v1.1.11/go.mod h1:83CECat5yLh5zVOf4P1ErAgKA5UDvKtgyUABdr3+MjI=
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=

utils/display.go

@@ -509,6 +509,7 @@ func FunctionsTable(funs []*meroxa.Function, hideHeaders bool) string {
 				{Align: simpletable.AlignCenter, Text: "OUTPUT STREAM"},
 				{Align: simpletable.AlignCenter, Text: "STATE"},
 				{Align: simpletable.AlignCenter, Text: "PIPELINE"},
+				{Align: simpletable.AlignCenter, Text: "APPLICATION"},
 			},
 		}
 	}
@@ -520,7 +521,8 @@ func FunctionsTable(funs []*meroxa.Function, hideHeaders bool) string {
 			{Align: simpletable.AlignCenter, Text: p.InputStream},
 			{Align: simpletable.AlignCenter, Text: p.OutputStream},
 			{Align: simpletable.AlignCenter, Text: p.Status.State},
-			{Align: simpletable.AlignCenter, Text: p.Pipeline.Name},
+			{Align: simpletable.AlignCenter, Text: p.Pipeline.Name.String},
+			{Align: simpletable.AlignCenter, Text: p.Application.Name.String},
 		}
 
 		table.Body.Cells = append(table.Body.Cells, r)
@@ -572,7 +574,11 @@ func FunctionTable(fun *meroxa.Function) string {
 		},
 		{
 			{Align: simpletable.AlignRight, Text: "Pipeline:"},
-			{Text: fun.Pipeline.Name},
+			{Text: fun.Pipeline.Name.String},
+		},
+		{
+			{Align: simpletable.AlignRight, Text: "Application:"},
+			{Text: fun.Application.Name.String},
 		},
 		{
 			{Align: simpletable.AlignRight, Text: "State:"},
@@ -827,7 +833,7 @@ func AppsTable(apps []*meroxa.Application, hideHeaders bool) string {
 	for _, app := range apps {
 		names := make([]string, 0)
 		for _, f := range app.Functions {
-			names = append(names, f.Name)
+			names = append(names, f.Name.String)
 		}
 		r := []*simpletable.Cell{
 			{Align: simpletable.AlignRight, Text: app.UUID},