Merge 60777d6 into 45bf2f0

VERSION

@@ -1 +1 @@
-0.1.2
+0.8.0

common/common.go

@@ -125,7 +125,7 @@ const (
 	// LabelGatewayEventSourceID is the label for gateway configuration ID
 	LabelGatewayEventSourceID = "event-source-id"
 
-	// Server Connection Timeout, 5 seconds
+	// Server Connection Timeout, 10 seconds
 	ServerConnTimeout = 10
 )
 

controllers/gateway/informer.go

@@ -13,6 +13,7 @@ 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 gateway
 
 import (

controllers/gateway/state.go

@@ -9,7 +9,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 )
 
-// persist the updates to the gateway resource
+// PersistUpdates of the gateway resource
 func PersistUpdates(client gwclient.Interface, gw *v1alpha1.Gateway, log *zerolog.Logger) (*v1alpha1.Gateway, error) {
 	gatewayClient := client.ArgoprojV1alpha1().Gateways(gw.ObjectMeta.Namespace)
 
@@ -23,7 +23,7 @@ func PersistUpdates(client gwclient.Interface, gw *v1alpha1.Gateway, log *zerolo
 			return oldgw, err
 		}
 		log.Info().Msg("re-applying updates on latest version and retrying update")
-		err = ReapplyUpdate(client, gw)
+		err = ReapplyUpdates(client, gw)
 		if err != nil {
 			log.Error().Err(err).Msg("failed to re-apply update")
 			return oldgw, err
@@ -33,8 +33,8 @@ func PersistUpdates(client gwclient.Interface, gw *v1alpha1.Gateway, log *zerolo
 	return gw, nil
 }
 
-// reapply the updates to gateway resource
-func ReapplyUpdate(client gwclient.Interface, gw *v1alpha1.Gateway) error {
+// ReapplyUpdates to gateway resource
+func ReapplyUpdates(client gwclient.Interface, gw *v1alpha1.Gateway) error {
 	return wait.ExponentialBackoff(common.DefaultRetry, func() (bool, error) {
 		gatewayClient := client.ArgoprojV1alpha1().Gateways(gw.Namespace)
 		g, err := gatewayClient.Update(gw)

gateways/common/doc.go

@@ -14,5 +14,5 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// This package contains structs and methods that are shared across different gateways.
+// Package common contains structs and methods that are shared across different gateways.
 package common

gateways/common/webhook.go

@@ -42,9 +42,10 @@ type Webhook struct {
 	ServerCertPath string `json:"serverCertPath,omitempty" protobuf:"bytes,4,opt,name=serverCertPath"`
 	// ServerKeyPath refers the file that contains private key
 	ServerKeyPath string `json:"serverKeyPath,omitempty" protobuf:"bytes,5,opt,name=serverKeyPath"`
+
 	// srv holds reference to http server
-	srv *http.Server   `json:"srv,omitempty"`
-	mux *http.ServeMux `json:"mux,omitempty"`
+	srv *http.Server
+	mux *http.ServeMux
 }
 
 // WebhookHelper is a helper struct
@@ -72,7 +73,7 @@ type activeServer struct {
 	errChan chan error
 }
 
-// RouteConfig contains configuration about a http route
+// RouteConfig contains configuration about an http route
 type RouteConfig struct {
 	Webhook            *Webhook
 	Configs            map[string]interface{}

gateways/community/aws-sns/config.go

@@ -17,18 +17,19 @@ limitations under the License.
 package aws_sns
 
 import (
+	"time"
+
 	"github.com/argoproj/argo-events/gateways/common"
 	"k8s.io/client-go/kubernetes"
-	"time"
 
 	"github.com/ghodss/yaml"
 	"github.com/rs/zerolog"
 	corev1 "k8s.io/api/core/v1"
 )
 
 const (
-	MESSAGE_TYPE_SUBSCRIPTION_CONFIRMATION = "SubscriptionConfirmation"
-	MESSAGE_TYPE_NOTIFICATION              = "Notification"
+	messageTypeSubscriptionConfirmation = "SubscriptionConfirmation"
+	messageTypeNotification             = "Notification"
 )
 
 var (

gateways/community/aws-sns/start.go

@@ -33,9 +33,9 @@ import (
 )
 
 const (
-	LabelSNSConfig       = "snsConfig"
-	LabelSNSSession      = "snsSession"
-	LabelSubscriptionArn = "subscriptionArn"
+	labelSNSConfig       = "snsConfig"
+	labelSNSSession      = "snsSession"
+	labelSubscriptionArn = "subscriptionArn"
 )
 
 var (
@@ -46,7 +46,7 @@ func init() {
 	go gwcommon.InitRouteChannels(helper)
 }
 
-// routeActiveHandler handles new route
+// RouteActiveHandler handles new routes
 func RouteActiveHandler(writer http.ResponseWriter, request *http.Request, rc *gwcommon.RouteConfig) {
 	var response string
 
@@ -76,11 +76,11 @@ func RouteActiveHandler(writer http.ResponseWriter, request *http.Request, rc *g
 		return
 	}
 
-	sc := rc.Configs[LabelSNSConfig].(*snsConfig)
+	sc := rc.Configs[labelSNSConfig].(*snsConfig)
 
 	switch snspayload.Type {
-	case MESSAGE_TYPE_SUBSCRIPTION_CONFIRMATION:
-		awsSession := rc.Configs[LabelSNSSession].(*snslib.SNS)
+	case messageTypeSubscriptionConfirmation:
+		awsSession := rc.Configs[labelSNSSession].(*snslib.SNS)
 		out, err := awsSession.ConfirmSubscription(&snslib.ConfirmSubscriptionInput{
 			TopicArn: &sc.TopicArn,
 			Token:    &snspayload.Token,
@@ -89,9 +89,9 @@ func RouteActiveHandler(writer http.ResponseWriter, request *http.Request, rc *g
 			logger.Error().Err(err).Msg("failed to send confirmation response to amazon")
 			return
 		}
-		rc.Configs[LabelSubscriptionArn] = out.SubscriptionArn
+		rc.Configs[labelSubscriptionArn] = out.SubscriptionArn
 
-	case MESSAGE_TYPE_NOTIFICATION:
+	case messageTypeNotification:
 		helper.ActiveEndpoints[rc.Webhook.Endpoint].DataCh <- body
 	}
 
@@ -100,11 +100,12 @@ func RouteActiveHandler(writer http.ResponseWriter, request *http.Request, rc *g
 	common.SendSuccessResponse(writer, response)
 }
 
+// PostActivate subscribes to the sns topic
 func (ese *SNSEventSourceExecutor) PostActivate(rc *gwcommon.RouteConfig) error {
 	logger := rc.Log.With().Str("event-source", rc.EventSource.Name).Str("endpoint", rc.Webhook.Endpoint).
 		Str("port", rc.Webhook.Port).Logger()
 
-	sc := rc.Configs[LabelSNSConfig].(*snsConfig)
+	sc := rc.Configs[labelSNSConfig].(*snsConfig)
 	// retrieve access key id and secret access key
 	accessKey, err := store.GetSecrets(ese.Clientset, ese.Namespace, sc.AccessKey.Name, sc.AccessKey.Key)
 	if err != nil {
@@ -135,7 +136,7 @@ func (ese *SNSEventSourceExecutor) PostActivate(rc *gwcommon.RouteConfig) error
 	}
 
 	snsSession := snslib.New(awsSession)
-	rc.Configs[LabelSNSSession] = snsSession
+	rc.Configs[labelSNSSession] = snsSession
 
 	logger.Info().Msg("subscribing to snsConfig topic")
 	if _, err := snsSession.Subscribe(&snslib.SubscribeInput{
@@ -150,19 +151,19 @@ func (ese *SNSEventSourceExecutor) PostActivate(rc *gwcommon.RouteConfig) error
 	return nil
 }
 
-// PostStop unsubscribes the topic
+// PostStop unsubscribes from the sns topic
 func PostStop(rc *gwcommon.RouteConfig) error {
-	awsSession := rc.Configs[LabelSNSSession].(*snslib.SNS)
+	awsSession := rc.Configs[labelSNSSession].(*snslib.SNS)
 	if _, err := awsSession.Unsubscribe(&snslib.UnsubscribeInput{
-		SubscriptionArn: rc.Configs[LabelSubscriptionArn].(*string),
+		SubscriptionArn: rc.Configs[labelSubscriptionArn].(*string),
 	}); err != nil {
 		rc.Log.Error().Err(err).Str("event-source-name", rc.EventSource.Name).Msg("failed to unsubscribe")
 		return err
 	}
 	return nil
 }
 
-// StartConfig runs a configuration
+// StartEventSource starts an SNS event source
 func (ese *SNSEventSourceExecutor) StartEventSource(eventSource *gateways.EventSource, eventStream gateways.Eventing_StartEventSourceServer) error {
 	defer gateways.Recover(eventSource.Name)
 
@@ -177,7 +178,7 @@ func (ese *SNSEventSourceExecutor) StartEventSource(eventSource *gateways.EventS
 	return gwcommon.ProcessRoute(&gwcommon.RouteConfig{
 		Webhook: sc.Hook,
 		Configs: map[string]interface{}{
-			LabelSNSConfig: sc,
+			labelSNSConfig: sc,
 		},
 		Log:                ese.Log,
 		EventSource:        eventSource,

gateways/community/gcp-pubsub/start.go

@@ -17,17 +17,13 @@ limitations under the License.
 package pubsub
 
 import (
-	"cloud.google.com/go/pubsub"
 	"context"
+
+	"cloud.google.com/go/pubsub"
 	"github.com/argoproj/argo-events/gateways"
 	"google.golang.org/api/option"
 )
 
-const (
-	// LabelGcpPubSubConfig is the label name of the GCP PubSub Config
-	LabelGcpPubSubConfig = "pubSubConfig"
-)
-
 // StartEventSource starts the GCP PubSub Gateway
 func (ese *GcpPubSubEventSourceExecutor) StartEventSource(eventSource *gateways.EventSource, eventStream gateways.Eventing_StartEventSourceServer) error {
 	defer gateways.Recover(eventSource.Name)

gateways/community/github/start.go

@@ -37,15 +37,15 @@ import (
 )
 
 const (
-	LabelGithubConfig = "config"
-	LabelGithubClient = "client"
-	LabelWebhook      = "hook"
+	labelGithubConfig = "config"
+	labelGithubClient = "client"
+	labelWebhook      = "hook"
 )
 
 const (
-	GithubSignatureHeader = "x-hub-signature"
-	GithubEventHeader     = "x-github-event"
-	GithubDeliveryHeader  = "x-github-delivery"
+	githubSignatureHeader = "x-hub-signature"
+	githubEventHeader     = "x-github-event"
+	githubDeliveryHeader  = "x-github-delivery"
 )
 
 var (
@@ -68,7 +68,7 @@ func (ese *GithubEventSourceExecutor) getCredentials(gs *corev1.SecretKeySelecto
 }
 
 func (ese *GithubEventSourceExecutor) PostActivate(rc *gwcommon.RouteConfig) error {
-	gc := rc.Configs[LabelGithubConfig].(*githubConfig)
+	gc := rc.Configs[labelGithubConfig].(*githubConfig)
 
 	c, err := ese.getCredentials(gc.APIToken)
 	if err != nil {
@@ -109,26 +109,27 @@ func (ese *GithubEventSourceExecutor) PostActivate(rc *gwcommon.RouteConfig) err
 	}
 
 	client := gh.NewClient(PATTransport.Client())
-	rc.Configs[LabelGithubClient] = client
-
-	ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
+	rc.Configs[labelGithubClient] = client
 
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
 	hook, _, err := client.Repositories.CreateHook(ctx, gc.Owner, gc.Repository, hookSetup)
 	if err != nil {
 		return fmt.Errorf("failed to create webhook. err: %+v", err)
 	}
-	rc.Configs[LabelWebhook] = hook
+	rc.Configs[labelWebhook] = hook
 
 	ese.Log.Info().Str("event-source-name", rc.EventSource.Name).Interface("hook-id", *hook.ID).Msg("github hook created")
 	return nil
 }
 
 func PostStop(rc *gwcommon.RouteConfig) error {
-	gc := rc.Configs[LabelGithubConfig].(*githubConfig)
-	client := rc.Configs[LabelGithubClient].(*gh.Client)
-	hook := rc.Configs[LabelWebhook].(*gh.Hook)
+	gc := rc.Configs[labelGithubConfig].(*githubConfig)
+	client := rc.Configs[labelGithubClient].(*gh.Client)
+	hook := rc.Configs[labelWebhook].(*gh.Hook)
 
-	ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
 	if _, err := client.Repositories.DeleteHook(ctx, gc.Owner, gc.Repository, *hook.ID); err != nil {
 		rc.Log.Error().Err(err).Str("event-source-name", rc.EventSource.Name).Msg("failed to delete github hook")
 		return err
@@ -151,7 +152,7 @@ func (ese *GithubEventSourceExecutor) StartEventSource(eventSource *gateways.Eve
 	return gwcommon.ProcessRoute(&gwcommon.RouteConfig{
 		Webhook: gc.Hook,
 		Configs: map[string]interface{}{
-			LabelGithubConfig: gc,
+			labelGithubConfig: gc,
 		},
 		Log:                ese.Log,
 		EventSource:        eventSource,
@@ -183,16 +184,16 @@ func verifySignature(secret []byte, signature string, body []byte) bool {
 }
 
 func validatePayload(secret []byte, headers http.Header, body []byte) error {
-	signature := headers.Get(GithubSignatureHeader)
+	signature := headers.Get(githubSignatureHeader)
 	if len(signature) == 0 {
 		return errors.New("no x-hub-signature header found")
 	}
 
-	if event := headers.Get(GithubEventHeader); len(event) == 0 {
+	if event := headers.Get(githubEventHeader); len(event) == 0 {
 		return errors.New("no x-github-event header found")
 	}
 
-	if id := headers.Get(GithubDeliveryHeader); len(id) == 0 {
+	if id := headers.Get(githubDeliveryHeader); len(id) == 0 {
 		return errors.New("no x-github-delivery header found")
 	}
 
@@ -225,7 +226,7 @@ func RouteActiveHandler(writer http.ResponseWriter, request *http.Request, rc *g
 		return
 	}
 
-	hook := rc.Configs[LabelWebhook].(*gh.Hook)
+	hook := rc.Configs[labelWebhook].(*gh.Hook)
 	if secret, ok := hook.Config["secret"]; ok {
 		if err := validatePayload([]byte(secret.(string)), request.Header, body); err != nil {
 			logger.Error().Err(err).Msg("request is not valid event notification")

gateways/community/gitlab/start.go

@@ -30,9 +30,9 @@ import (
 )
 
 const (
-	LabelGitlabConfig = "config"
-	LabelGitlabClient = "client"
-	LabelWebhook      = "hook"
+	labelGitlabConfig = "config"
+	labelGitlabClient = "client"
+	labelWebhook      = "hook"
 )
 
 var (
@@ -55,7 +55,7 @@ func (ese *GitlabEventSourceExecutor) getCredentials(gs *GitlabSecret) (*cred, e
 }
 
 func (ese *GitlabEventSourceExecutor) PostActivate(rc *gwcommon.RouteConfig) error {
-	gl := rc.Configs[LabelGitlabConfig].(*glab)
+	gl := rc.Configs[labelGitlabConfig].(*glab)
 
 	c, err := ese.getCredentials(gl.AccessToken)
 	if err != nil {
@@ -67,7 +67,7 @@ func (ese *GitlabEventSourceExecutor) PostActivate(rc *gwcommon.RouteConfig) err
 		return fmt.Errorf("failed to set gitlab base url, err: %+v", err)
 	}
 
-	rc.Configs[LabelGitlabClient] = client
+	rc.Configs[labelGitlabClient] = client
 
 	formattedUrl := gwcommon.GenerateFormattedURL(gl.Hook)
 
@@ -90,16 +90,16 @@ func (ese *GitlabEventSourceExecutor) PostActivate(rc *gwcommon.RouteConfig) err
 	if err != nil {
 		return fmt.Errorf("failed to add project hook. err: %+v", err)
 	}
-	rc.Configs[LabelWebhook] = hook
+	rc.Configs[labelWebhook] = hook
 
 	rc.Log.Info().Str("event-source-name", rc.EventSource.Name).Interface("hook-id", hook.ID).Msg("gitlab hook created")
 	return nil
 }
 
 func PostStop(rc *gwcommon.RouteConfig) error {
-	gl := rc.Configs[LabelGitlabConfig].(*glab)
-	client := rc.Configs[LabelGitlabClient].(*gitlab.Client)
-	hook := rc.Configs[LabelWebhook].(*gitlab.ProjectHook)
+	gl := rc.Configs[labelGitlabConfig].(*glab)
+	client := rc.Configs[labelGitlabClient].(*gitlab.Client)
+	hook := rc.Configs[labelWebhook].(*gitlab.ProjectHook)
 
 	if _, err := client.Projects.DeleteProjectHook(gl.ProjectId, hook.ID); err != nil {
 		rc.Log.Error().Err(err).Str("event-source-name", rc.EventSource.Name).Interface("hook-id", hook.ID).Msg("failed to delete gitlab hook")
@@ -153,7 +153,7 @@ func (ese *GitlabEventSourceExecutor) StartEventSource(eventSource *gateways.Eve
 	return gwcommon.ProcessRoute(&gwcommon.RouteConfig{
 		Webhook: gl.Hook,
 		Configs: map[string]interface{}{
-			LabelGitlabConfig: gl,
+			labelGitlabConfig: gl,
 		},
 		Log:                ese.Log,
 		EventSource:        eventSource,