AliECS core Kafka producer

Makefile

@@ -68,8 +68,7 @@ WHAT_o2-apricot_BUILD_FLAGS=$(BUILD_ENV_FLAGS)
 
 INSTALL_WHAT:=$(patsubst %, install_%, $(WHAT))
 
-
-GENERATE_DIRS := ./apricot ./coconut/cmd ./common/runtype ./common/system ./core ./core/integration/ccdb ./core/integration/dcs ./core/integration/ddsched ./core/integration/kafka ./core/integration/odc ./executor ./walnut ./core/integration/trg ./core/integration/bookkeeping
+GENERATE_DIRS := ./apricot ./coconut/cmd ./common ./common/runtype ./common/system ./core ./core/integration/ccdb ./core/integration/dcs ./core/integration/ddsched ./core/integration/kafka ./core/integration/odc ./executor ./walnut ./core/integration/trg ./core/integration/bookkeeping
 SRC_DIRS := ./apricot ./cmd/* ./core ./coconut ./executor ./common ./configuration ./occ/peanut ./walnut
 TEST_DIRS := ./configuration/cfgbackend ./configuration/componentcfg
 GO_TEST_DIRS := ./core/repos ./core/integration/dcs
@@ -87,7 +86,7 @@ HAS_PROTOC := $(shell command -v $(GOPROTOCPATH) 2> /dev/null)
 .EXPORT_ALL_VARIABLES:
 CGO_ENABLED = 0
 
-.PHONY: build all install generate test debugtest vet fmt clean cleanall help $(WHAT) tools vendor doc docs
+.PHONY: build all install generate test debugtest vet fmt clean cleanall help $(WHAT) tools vendor doc docs fdset
 
 build: $(WHAT)
 
@@ -274,6 +273,15 @@ tools/protoc:
 
 docs: docs/coconut docs/grpc docs/swaggo
 
+fdset:
+	@echo -e "building fdset files  \033[1;33m==>\033[0m  \033[1;34m./common/protos\033[0m"
+
+	@mkdir -p fdset
+	@cd common/protos && protoc -o events.fdset events.proto && cd ../..
+	@mv common/protos/events.fdset fdset
+
+	@echo -e "to consume with \033[1;33mhttps://github.com/sevagh/pq\033[0m:  FDSET_PATH=./fdset pq kafka aliecs.environment --brokers kafka-broker-hostname:9092 --beginning --msgtype events.Event"
+
 docs/coconut:
 	@echo -e "generating coconut documentation  \033[1;33m==>\033[0m  \033[1;34m./coconut/doc\033[0m"
 	@cd coconut/doc && go run . && cd ../..

coconut/cmd/environment.go

@@ -22,7 +22,7 @@
  * Intergovernmental Organization or submit itself to any jurisdiction.
  */
 
-//go:generate protoc -I ../../core --go_out=.. --go-grpc_out=.. --go_opt=paths=source_relative --go-grpc_opt=paths=source_relative protos/o2control.proto
+//go:generate protoc -I=../../core -I=../../common --go_out=.. --go_opt=paths=source_relative --go-grpc_opt=paths=source_relative --go-grpc_out=require_unimplemented_servers=false:.. protos/o2control.proto
 
 package cmd
 

coconut/cmd/environment_create.go

@@ -66,6 +66,7 @@ func init() {
 
 	environmentCreateCmd.Flags().BoolP("auto", "a", false, "create an autorun environment")
 	environmentCreateCmd.Flags().BoolP("public", "p", true, "control public rights of the environment")
+	environmentCreateCmd.Flags().BoolP("asynchronous", "y", false, "use asynchronous mode for environment creation")
 
 	environmentCreateCmd.Flags().StringP("extra-vars", "e", "", "values passed using key=value CSV or JSON syntax, interpreted as strings `key1=val1,key2=val2` or `{\"key1\": \"value1\", \"key2\": \"value2\"}`")
 }

coconut/cmd/repo.go

@@ -22,8 +22,6 @@
  * Intergovernmental Organization or submit itself to any jurisdiction.
  */
 
-//go:generate protoc -I ../../core --go_out=.. --go-grpc_out=.. --go_opt=paths=source_relative --go-grpc_opt=paths=source_relative protos/o2control.proto
-
 package cmd
 
 import (

coconut/control/control.go

@@ -420,8 +420,14 @@ func CreateEnvironment(cxt context.Context, rpc *coconut.RpcClient, cmd *cobra.C
 	// TODO: add support for setting visibility here OCTRL-178
 	// TODO: add support for acquiring bot config here OCTRL-177
 
+	asynchronous, _ := cmd.Flags().GetBool("asynchronous")
+
 	var response *pb.NewEnvironmentReply
-	response, err = rpc.NewEnvironment(cxt, &pb.NewEnvironmentRequest{WorkflowTemplate: wfPath, Vars: extraVarsMap, Public: public}, grpc.EmptyCallOption{})
+	if asynchronous {
+		response, err = rpc.NewEnvironmentAsync(cxt, &pb.NewEnvironmentRequest{WorkflowTemplate: wfPath, Vars: extraVarsMap, Public: public}, grpc.EmptyCallOption{})
+	} else {
+		response, err = rpc.NewEnvironment(cxt, &pb.NewEnvironmentRequest{WorkflowTemplate: wfPath, Vars: extraVarsMap, Public: public}, grpc.EmptyCallOption{})
+	}
 	if err != nil {
 		return
 	}