re-architecting around gateways and sensors #92

VaibhavPage · 2018-08-22T17:43:22Z

No description provided.

…ing, making sensor repeatable

CLAassistant · 2018-08-22T17:43:29Z

All committers have signed the CLA.

magaldima · 2018-08-25T14:55:33Z

Couple things:

can you provide a brief summary of the high-level API changes involved in this PR?
please update the readme's, quickstart and docs with this new information.
please update the hack/k8s/manifests with the new yaml files for gateway-deployment.
the pkg/*-client directories can be changed to just pkg/clients/ with sub directories for gateway and sensor just like the pkg/apis/ directory.
can we group the top level gateway-controller and sensor-controller dirs under a common controllers directory and rename the sub dirs without the -controller suffix?
we should remove all project dependencies and reference to github.com/micro/go-micro and github.com/micro/go-plugins
we should be consistent with the logger we use across the project. the gateway controller, transformer, and individual gateways use github.com/rs/zerolog while the sensor controller uses github.com/sirupsen/logrus. What's your reasoning for using this new logger?

Overall, this looks really cool, I'm excited to try this out.

magaldima · 2018-08-25T14:56:32Z

Makefile

-DOCKER_PUSH=false
-IMAGE_NAMESPACE=argoproj
+DOCKER_PUSH=true
+IMAGE_NAMESPACE=metalgearsolid


why this name?

its my registry name :) . The changes such as registry name, adding namespaces to manifests etc were all part of local testing and I had planned to remove them before resolving the WIP status.

common/common.go

magaldima · 2018-08-25T14:59:13Z

common/common.go

+	GatewayTransformerConfigMapEnvVar = "GATEWAY_TRANSFORMER_CONFIG_MAP"
+
+	// GatewayEventTransformerImage is image for gateway event transformer
+	GatewayEventTransformerImage = "metalgearsolid/gateway-transformer"


see above comment for namespace

magaldima · 2018-08-25T15:00:21Z

common/util.go

+
+// SendSuccessResponse sends http success response
+func SendSuccessResponse(writer *http.ResponseWriter) {
+	(*writer).WriteHeader(http.StatusOK)


why do you need to dereference the writer here?

examples/gateways/artifact-gateway-configmap.yaml

magaldima · 2018-08-25T15:13:39Z

hack/k8s/manifests/gateway-controller-configmap.yaml

+kind: ConfigMap
+metadata:
+  name: gateway-controller-configmap
+  namespace: argo-events


let's remove the namespace references from manifests..

magaldima · 2018-08-25T15:14:37Z

hack/update-codegen.sh

 bash -x ${CODEGEN_PKG}/generate-groups.sh "deepcopy,client,informer,lister" \
-  github.com/argoproj/argo-events/pkg/client github.com/argoproj/argo-events/pkg/apis \
+  github.com/argoproj/argo-events/pkg/sensor-client github.com/argoproj/argo-events/pkg/apis \


we should be using github.com/argoproj/argo-events/pkg/client for both

magaldima · 2018-08-25T15:16:18Z

pkg/apis/gateway/v1alpha1/types.go

+type Gateway struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata" protobuf:"bytes,1,opt,name=metadata"`
+	Status            NodePhase   `json:"type" protobuf:"bytes,2,opt,name=type"`


should this be of type GatewayStatus instead? make sure to edit the annotations to match the name as well!

magaldima · 2018-08-25T15:17:11Z

pkg/apis/gateway/v1alpha1/types.go

+	Sensors []string `json:"sensors" protobuf:"bytes,8,opt,name=sensors"`
+
+	// ServiceAccountName is name of service account to run the gateway
+	ServiceAccountName string `json:"serviceAccountName" protobuf:"bytes,9,opt,name=service_account_name"`


the protobuf naming scheme should be camelCase: serviceAccountName

magaldima · 2018-08-25T15:17:37Z

pkg/apis/sensor/v1alpha1/types.go

@@ -25,18 +25,6 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1"
 )

-// SignalType is the type of the signal
-type SignalType string


great to see this go!

magaldima

package structure much better!
only major concern is the introduction of the registered* map in the configuration of each gateway. Should these store state about the Gateways? What happens when a Gateway goes down?

magaldima · 2018-08-27T23:21:41Z

controllers/gateway/operator.go

 	case v1alpha1.NodePhaseNew:
+
+		goc.markGatewayPhase(v1alpha1.NodePhaseNew, "new")


this seems redundant as the phase is already new

oops. got carried away there marking gateway phases :)

magaldima · 2018-08-27T23:22:21Z

controllers/gateway/operator.go

 		// Update node phase to running
-		goc.gw.Status = v1alpha1.NodePhaseRunning
+		goc.gw.Status.Phase = v1alpha1.NodePhaseRunning


un-needed, you update the phase at the end of operation and this update happens in the markGatewayPhase function anyway..

controllers/gateway/operator.go

magaldima · 2018-08-27T23:24:11Z

controllers/gateway/operator.go

-					return err
-				}
-			}
+			goc.markGatewayPhase(v1alpha1.NodePhaseError, "failed to create transformer gateway configuration")


it's useful to include the err in the message.

magaldima · 2018-08-27T23:25:28Z

controllers/gateway/operator.go

@@ -159,22 +168,13 @@ func (goc *gwOperationCtx) operate() error {
 		_, err = goc.controller.kubeClientset.AppsV1().Deployments(goc.gw.Namespace).Create(gatewayDeployment)
 		if err != nil {
 			goc.log.Error().Err(err).Msg("failed gateway deployment")
-			goc.gw.Status = v1alpha1.NodePhaseError
+			goc.markGatewayPhase(v1alpha1.NodePhaseError, "failed gateway deployment")


same deal with including err in the message

magaldima · 2018-08-27T23:27:35Z

controllers/gateway/operator.go

 		} else {
-			goc.gw.Status = v1alpha1.NodePhaseRunning
+			// expose gateway if service is configured
 			if goc.gw.Spec.Service.Port != 0 {
 				goc.createGatewayService()


why do we fail silently here?

As the service is not the absolutely critical part of the gateway. An user can create the service manually in the event of service creation failure.

do we have a status phase to represent this possible condition state? if not, we should so when the user investigates why stuff is not working, he knows how to resolve.

magaldima · 2018-08-27T23:31:28Z

gateways/core/artifact/s3.go

-	registeredArtifacts []S3Artifact
+	gatewayConfig       *gateways.GatewayConfig
+	// registeredArtifacts contains map of registered s3 artifacts
+	registeredArtifacts map[uint64]*S3Artifact


what is the purpose of this map? it doesn't look like we're saving anything in this? also the gateway doesn't want to be storing state correct?

It won't be storing the state of the gateway but rather the configurations. A gateway can be configured with multiple configurations. Each configuration is run in different go routine and produces the events. The idea is to use a single gateway among multiple use cases that differ only in configurations.

We are watching gateway-configmap to dynamically add new configuration and the purpose of this map is to keep track of existing configurations of gateway. Whenever an user updates the configmap with new configuration, this map is used to first check whether the configuration already exists or is it a completely new configuration.

so what happens when a gateway goes down and loses this information? does it pick up the existing configurations from the configMap and restart the necessary go routines?

VaibhavPage · 2018-08-28T09:56:57Z

It makes sense to add number of completions count in sensor #6

VaibhavPage · 2018-08-28T19:59:14Z

I was looking for a fast JSON output logger and came across zerolog. From the benchmarking posted on their github, it performs better than uber/zap.

VaibhavPage · 2018-08-28T22:57:26Z

controllers/gateway/validate.go

+	if len(goc.gw.Spec.Sensors) <= 0 {
+		return fmt.Errorf("no associated sensor with gateway")
+	}
+	if goc.gw.Spec.ServiceAccountName == "" {


use default service account instead

VaibhavPage · 2018-08-28T22:58:17Z

controllers/gateway/operator.go

+	}
+}
+
+// persist the updates to the Sensor resource


update comment, remove sensor resource

controllers/sensor/controller.go

…ained using k8 events

VaibhavPage · 2018-09-11T21:38:03Z

Added K8 events for gateway.

VaibhavPage · 2018-09-11T22:54:41Z

K8 events currently work as mechanism for gateway to update it's state and avoid synchronization within multiple go routines altogether and remove need to have multiple gateway resource watchers. Event watcher currently sit within gateway processor. I am thinking of moving it to Gateway controller. Event watcher in gateway processor listens to k8 events that are specific to that gateway, but Gateway operator will listen to k8 events of type Gateway and not specific set of events. Also gateway controller will need to fetch corresponding gateway resource every time there is a new Gateway event. Currently we don't fetch gateway resource every time there is new event, because gateway has only one gateway resource to watch and it updates the same. Thats the trade-off between having a separate watcher for each gateway and having just one watcher at Gateway controller.

I am also thinking of watching k8 events to maintain Sensor state at Sensor controller. This will remove necessity to watch sensor resource in sensor pod. Same trade-off applies here as well.

…ding support for NATS and Kafka dispatch mechanism

magaldima

really cool stuff. I put some questions / comments here, but looking good! thanks for the documentation and diagrams!

magaldima · 2018-09-12T23:58:07Z

common/util.go

@@ -90,3 +92,21 @@ func SendErrorResponse(writer http.ResponseWriter) {
 	writer.WriteHeader(http.StatusBadRequest)
 	writer.Write([]byte(ErrorResponse))
 }
+
+// RandomStringGenerator generates a random string
+func RandomStringGenerator() string {


is there not a library out there to do this?

magaldima · 2018-09-13T00:27:23Z

controllers/gateway/transform/transformer.go

 	return ce, nil
 }

-// dispatches the event to configured sensor
+// getWatcherIP returns IP of service which backs given component.
+func (toc *tOperationCtx) getWatcherIP(name string) (string, string, error) {


this method seems a bit odd to me.. shouldn't we be using DNS with service names instead of inspected for IP addresses?

magaldima · 2018-09-13T00:31:45Z

docs/custom-gateway.md

+
+For detailed implementation, check out [Calendar HTTP gateway](https://github.com/argoproj/argo-events/tree/eventing/gateways/rest/calendar)
+
+###### Advantage of writing a gRPC or http gateway is that you can use other languages for implementation of gateway processor server.   


isn't this an advantage for all types of gateways? as long as the gateway processor server has a library to interface with the gateway client.. be that NATS, Kafka, HTTP. you just need a supported protocol + infra (if needed in case of streams)

…e whenever configuration is removed from gateway config map. remaned workflows in example sensors.

magaldima · 2018-09-16T19:05:41Z

common/k8-events.go

+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: event.Namespace,
+			Name:      event.Name + "-" + RandomStringGenerator(),


can you not just use GenerateName here instead?

yup, had it changed in my local. I'll push some new changes with test cases

magaldima · 2018-09-16T19:09:35Z

controllers/sensor/signal-filter.go

+
+		if timeFilter.Start != "" && timeFilter.Stop != "" {
+			se.log.Info().Str("start time format", currentTStr + " " + timeFilter.Start).Msg("start time format")
+			startTime, err := time.Parse("2006-01-02 15:04:05", currentTStr + " " + timeFilter.Start)


let's make the format string 2006-01-02 15:04:05 a constant since we use it more than once and seeing this may be confusing for some.

…ents for sensor

…ndler

VaibhavPage · 2018-09-18T05:11:11Z

removing WIP status

VaibhavPage · 2018-09-22T00:04:51Z

ready to merge

magaldima

Changes look good! I'm going to create a follow-up ticket to specifically address writing better unit tests and integration tests.

Good work on this @VaibhavPage ! Excited to get this out.

magaldima · 2018-09-23T22:20:33Z

controllers/gateway/transform/transformer.go

@@ -100,12 +100,12 @@ func (toc *tOperationCtx) transform(r *http.Request) (*sv1alpha.Event, error) {
 		Context: sv1alpha.EventContext{
 			CloudEventsVersion: common.CloudEventsVersion,
 			EventID:            fmt.Sprintf("%x", eventId),
-			ContentType:        r.Header.Get(common.HeaderContentType),
-			EventTime:          metav1.Time{Time: time.Now().UTC()},
+			ContentType:        "application/json",


how do you know this to always be true?

magaldima · 2018-09-23T22:21:22Z

controllers/sensor/controller.go

@@ -88,7 +90,7 @@ func (c *SensorController) processNextItem() bool {

 	obj, exists, err := c.informer.GetIndexer().GetByKey(key.(string))
 	if err != nil {
-		log.Warnf("failed to get sensor '%s' from informer index: %+v", key, err)
+		fmt.Errorf("failed to get sensor '%s' from informer index: %+v", key, err)


should we be using a logger instead of just printing?

magaldima · 2018-10-01T01:13:55Z

controllers/gateway/validate_test.go

@@ -10,6 +10,6 @@ func TestGwOperationCtx_validate(t *testing.T) {
 	gateway, err := getGateway()
 	assert.Nil(t, err)
 	goc := newGatewayOperationCtx(gateway, fakeController)
-	err = goc.validate()
+	err = Validate(goc.gw)
 	assert.Nil(t, err)


we should probably beef up this unit tests at some point later.. we can make a note of it and address in a future ticket.

* added gateway for processing outside events, making gateway long running, making sensor repeatable * refactoring k8s manifests * adding s3 gateway * making s3 and webhook gateway mult configurable * refactor of calendar, s3 and webhook gateway done. started with streams * added nats gateway * renamed signals to gateways. moved out-of-box gateways to core. * refactor gateways * removing signals folder * added kafka gateway * gateway and sensor dirs under a common controllers and clients directory * adding resource, amqp and mqtt gateways. refactored sensor watcher * adding gateway examples, readme etc * adding gateway docs. refactor gateway and sensor examples * added watcher for sensor. Completion count for sensor. Updated gateway doc * sensor's signal are gateway-name/configuration-name * making gateway processor as gRPC server * giving user options to implement gateway either by mimicing core gateways, a grpc server or completely custom implementation * gateway spec changed. divided extending gateways into 3 parts. started updating gateway examples * updated gateway examples with deploySpec. started writing custom gateway doc. * adding wiki for writing custom gateways * updating wiki. grpc gateway updated DialOptions * added REST gateways. wiki to follow * refactor rest gateways * added k8 events for gateway. Gateway configuration state is now maintained using k8 events * minor refactor in k8 events * added gateways as watchers for gateway along with sensors. started adding support for NATS and Kafka dispatch mechanism * added k8 event for escalation. nodes are deleted from gateway resource whenever configuration is removed from gateway config map. remaned workflows in example sensors. * added unit tests. refactored sensor notification handler. added k8 events for sensor * adding test cases for sensor controller, operator and notification handler * updating webhook gateway * fixed issue with syncing gateway nodes and gateway configmap * test: cleaning up test cases * update docker registry to argoproj * exposing gateway and sensor validate method * update dependencies

VaibhavPage added 2 commits August 22, 2018 10:48

added gateway for processing outside events, making gateway long runn…

12af598

…ing, making sensor repeatable

refactoring k8s manifests

4bb4457

VaibhavPage requested review from plasma86, shrinandj, spk83 and magaldima August 22, 2018 17:43

VaibhavPage mentioned this pull request Aug 22, 2018

evalulate re-architecting around knative-eventing #79

Closed

VaibhavPage added 8 commits August 22, 2018 19:05

adding s3 gateway

95c2548

making s3 and webhook gateway mult configurable

5839ee9

refactor of calendar, s3 and webhook gateway done. started with streams

c3542da

added nats gateway

543f698

renamed signals to gateways. moved out-of-box gateways to core.

44956f4

refactor gateways

78843be

removing signals folder

cdc20f3

added kafka gateway

2ab9407

magaldima suggested changes Aug 25, 2018

View reviewed changes

gateway and sensor dirs under a common controllers and clients directory

7caee70

magaldima suggested changes Aug 27, 2018

View reviewed changes

magaldima mentioned this pull request Aug 27, 2018

Signal Stream type amqp does not work #90

Closed

This was referenced Aug 28, 2018

Calendar sensor with Schedule: @every 30s triggers only once #91

Closed

repeatable sensors #6

Closed

adding resource, amqp and mqtt gateways. refactored sensor watcher

2db02fe

adding gateway examples, readme etc

62ccce9

VaibhavPage commented Aug 28, 2018

View reviewed changes

controllers/sensor/controller.go Show resolved Hide resolved

VaibhavPage added 5 commits September 6, 2018 15:49

adding wiki for writing custom gateways

11110d3

updating wiki. grpc gateway updated DialOptions

8947fe5

added REST gateways. wiki to follow

6828b2e

refactor rest gateways

50ec67c

added k8 events for gateway. Gateway configuration state is now maint…

b8c79c0

…ained using k8 events

VaibhavPage added 2 commits September 11, 2018 19:05

minor refactor in k8 events

99b04d5

added gateways as watchers for gateway along with sensors. started ad…

655028a

…ding support for NATS and Kafka dispatch mechanism

magaldima reviewed Sep 13, 2018

View reviewed changes

added k8 event for escalation. nodes are deleted from gateway resourc…

3b0c7b0

…e whenever configuration is removed from gateway config map. remaned workflows in example sensors.

magaldima reviewed Sep 16, 2018

View reviewed changes

VaibhavPage added 3 commits September 17, 2018 02:54

added unit tests. refactored sensor notification handler. added k8 ev…

b43c934

…ents for sensor

adding test cases for sensor controller, operator and notification ha…

970c63b

…ndler

updating webhook gateway

9ec8c44

VaibhavPage changed the title ~~WIP: re-architecting around gateways and sensors~~ re-architecting around gateways and sensors Sep 18, 2018

VaibhavPage added 4 commits September 20, 2018 19:27

fixed issue with syncing gateway nodes and gateway configmap

6433b77

test: cleaning up test cases

6691b07

updating dependencies

d08ac54

update docker registry to argoproj

92b31a5

ilyasotkov mentioned this pull request Sep 23, 2018

[WIP] Update argo-events chart for the upcoming v0.5 release (argoproj/argo-events#92) argoproj/argo-helm#21

Closed

5 tasks

VaibhavPage added 2 commits September 27, 2018 12:30

exposing gateway and sensor validate method

ac90e6b

update dependencies

edfa3a1

magaldima approved these changes Oct 1, 2018

View reviewed changes

magaldima merged commit 3272f1f into master Oct 1, 2018

VaibhavPage mentioned this pull request Oct 5, 2018

NPE in artifact s3 #93

Closed

VaibhavPage deleted the eventing branch October 28, 2018 12:01

		case v1alpha1.NodePhaseNew:

		goc.markGatewayPhase(v1alpha1.NodePhaseNew, "new")


		For detailed implementation, check out [Calendar HTTP gateway](https://github.com/argoproj/argo-events/tree/eventing/gateways/rest/calendar)

		###### Advantage of writing a gRPC or http gateway is that you can use other languages for implementation of gateway processor server.

re-architecting around gateways and sensors #92

re-architecting around gateways and sensors #92

Conversation

VaibhavPage commented Aug 22, 2018

CLAassistant commented Aug 22, 2018 • edited

magaldima commented Aug 25, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

magaldima left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

VaibhavPage Aug 28, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

VaibhavPage Aug 28, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

VaibhavPage commented Aug 28, 2018

VaibhavPage commented Aug 28, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

VaibhavPage commented Sep 11, 2018

VaibhavPage commented Sep 11, 2018 • edited

magaldima left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

VaibhavPage Sep 16, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

VaibhavPage commented Sep 18, 2018

VaibhavPage commented Sep 22, 2018

magaldima left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

CLAassistant commented Aug 22, 2018 •

edited

VaibhavPage Aug 28, 2018 •

edited

VaibhavPage Aug 28, 2018 •

edited

VaibhavPage commented Sep 11, 2018 •

edited

VaibhavPage Sep 16, 2018 •

edited