feat: workflow status listener - RK-19190 (#85)

* feat: workflow status listener Signed-off-by: Gosha <[email protected]> * fix: update file struct Signed-off-by: Gosha <[email protected]> * fix: change naming Signed-off-by: Gosha <[email protected]> * fix: add github status Signed-off-by: Gosha <[email protected]> * fix: create github notifier Signed-off-by: Gosha <[email protected]> * fix: handler removed Signed-off-by: Gosha <[email protected]> * fix: e2e test to use argo right address Signed-off-by: Gosha <[email protected]> * fix: github status norifier Signed-off-by: Gosha <[email protected]> * feat: update piper notify status Signed-off-by: Gosha <[email protected]> * feat: move event handler to diffrent interface Signed-off-by: Gosha <[email protected]> * feat: change panic to info Signed-off-by: Gosha <[email protected]> * fix: remove unnecessary label Signed-off-by: Gosha <[email protected]> * fix: seprate workflow handler to diffrent file Signed-off-by: Gosha <[email protected]> * fix: added tests Signed-off-by: Gosha <[email protected]> * fix: fixed some test issues Signed-off-by: Gosha <[email protected]> * fix: chnage piper/notify label Signed-off-by: Gosha <[email protected]> * fix: chnage piper/notify label Signed-off-by: Gosha <[email protected]> * fix: code review changes Signed-off-by: Gosha <[email protected]> * fix: mockGitProvider implementation mismatch Signed-off-by: Gosha <[email protected]> --------- Signed-off-by: Gosha <[email protected]> Signed-off-by: GoshaDozoretz <[email protected]>
Rookout · Jul 8, 2023 · a8a470e · a8a470e
1 parent 9beb0e7
commit a8a470e
Show file tree

Hide file tree

Showing 16 changed files with 729 additions and 45 deletions.
diff --git a/.github/workflows/e2e.yaml b/.github/workflows/e2e.yaml
@@ -82,6 +82,7 @@ jobs:
           --set piper.gitProvider.webhook.repoList={piper-e2e-test} \
           --set piper.gitProvider.organization.name="rookout" \
           --set image.repository=localhost:5001 \
+          --set piper.argoWorkflows.server.address="${{ env.NGROK_URL }}/argo" \
           --set env\[0\].name=ROOKOUT_CONTROLLER_HOST,env\[0\].value=wss://dogfood.control.rookout.com && \
           sleep 20 && kubectl logs deployment/piper
           kubectl wait  \

diff --git a/cmd/piper/piper.go b/cmd/piper/piper.go
@@ -1,16 +1,15 @@
 package main
 
 import (
-	"log"
-
+	rookout "github.com/Rookout/GoSDK"
 	"github.com/rookout/piper/pkg/clients"
 	"github.com/rookout/piper/pkg/conf"
+	"github.com/rookout/piper/pkg/event_handler"
 	"github.com/rookout/piper/pkg/git_provider"
 	"github.com/rookout/piper/pkg/server"
 	"github.com/rookout/piper/pkg/utils"
 	workflowHandler "github.com/rookout/piper/pkg/workflow_handler"
-
-	rookout "github.com/Rookout/GoSDK"
+	"log"
 )
 
 func main() {
@@ -46,5 +45,11 @@ func main() {
 		Workflows:   workflows,
 	}
 
+	err = globalClients.GitProvider.SetWebhook()
+	if err != nil {
+		panic(err)
+	}
+
+	event_handler.Start(cfg, globalClients)
 	server.Start(cfg, globalClients)
 }
diff --git a/pkg/event_handler/github_event_notifier.go b/pkg/event_handler/github_event_notifier.go
@@ -0,0 +1,58 @@
+package event_handler
+
+import (
+	"context"
+	"fmt"
+	"github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
+	"github.com/rookout/piper/pkg/clients"
+	"github.com/rookout/piper/pkg/conf"
+)
+
+var workflowTranslationToGithubMap = map[string]string{
+	"":          "pending",
+	"Pending":   "pending",
+	"Running":   "pending",
+	"Succeeded": "success",
+	"Failed":    "failure",
+	"Error":     "error",
+}
+
+type githubNotifier struct {
+	cfg     *conf.GlobalConfig
+	clients *clients.Clients
+}
+
+func NewGithubEventNotifier(cfg *conf.GlobalConfig, clients *clients.Clients) EventNotifier {
+	return &githubNotifier{
+		cfg:     cfg,
+		clients: clients,
+	}
+}
+
+func (gn *githubNotifier) Notify(ctx *context.Context, workflow *v1alpha1.Workflow) error {
+	fmt.Printf("Notifing workflow, %s\n", workflow.GetName())
+
+	repo, ok := workflow.GetLabels()["repo"]
+	if !ok {
+		return fmt.Errorf("failed get repo label for workflow: %s", workflow.GetName())
+	}
+	commit, ok := workflow.GetLabels()["commit"]
+	if !ok {
+		return fmt.Errorf("failed get commit label for workflow: %s", workflow.GetName())
+	}
+
+	workflowLink := fmt.Sprintf("%s/workflows/%s/%s", gn.cfg.WorkflowServerConfig.ArgoAddress, gn.cfg.Namespace, workflow.GetName())
+
+	status, ok := workflowTranslationToGithubMap[string(workflow.Status.Phase)]
+	if !ok {
+		return fmt.Errorf("failed to translate workflow status to github stasuts for %s status: %s", workflow.GetName(), workflow.Status.Phase)
+	}
+
+	message := workflow.Status.Message
+	err := gn.clients.GitProvider.SetStatus(ctx, &repo, &commit, &workflowLink, &status, &message)
+	if err != nil {
+		return fmt.Errorf("failed to set status for workflow %s: %s", workflow.GetName(), err)
+	}
+
+	return nil
+}
diff --git a/pkg/event_handler/github_event_notifier_test.go b/pkg/event_handler/github_event_notifier_test.go
@@ -0,0 +1,208 @@
+package event_handler
+
+import (
+	"context"
+	"errors"
+	"github.com/rookout/piper/pkg/git_provider"
+	assertion "github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"net/http"
+	"testing"
+
+	"github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
+	"github.com/rookout/piper/pkg/clients"
+	"github.com/rookout/piper/pkg/conf"
+)
+
+type mockGitProvider struct{}
+
+func (m *mockGitProvider) GetFile(ctx *context.Context, repo string, branch string, path string) (*git_provider.CommitFile, error) {
+	return nil, nil
+}
+
+func (m *mockGitProvider) GetFiles(ctx *context.Context, repo string, branch string, paths []string) ([]*git_provider.CommitFile, error) {
+	return nil, nil
+}
+
+func (m *mockGitProvider) ListFiles(ctx *context.Context, repo string, branch string, path string) ([]string, error) {
+	return nil, nil
+}
+
+func (m *mockGitProvider) SetWebhook() error {
+	return nil
+}
+
+func (m *mockGitProvider) UnsetWebhook(ctx *context.Context) error {
+	return nil
+}
+
+func (m *mockGitProvider) HandlePayload(request *http.Request, secret []byte) (*git_provider.WebhookPayload, error) {
+	return nil, nil
+}
+
+func (m *mockGitProvider) SetStatus(ctx *context.Context, repo *string, commit *string, linkURL *string, status *string, message *string) error {
+	return nil
+}
+
+func TestNotify(t *testing.T) {
+	assert := assertion.New(t)
+	ctx := context.Background()
+
+	// Define test cases
+	tests := []struct {
+		name        string
+		workflow    *v1alpha1.Workflow
+		wantedError error
+	}{
+		{
+			name: "Succeeded workflow",
+			workflow: &v1alpha1.Workflow{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-workflow",
+					Labels: map[string]string{
+						"repo":   "test-repo",
+						"commit": "test-commit",
+					},
+				},
+				Status: v1alpha1.WorkflowStatus{
+					Phase:   v1alpha1.WorkflowSucceeded,
+					Message: "",
+				},
+			},
+			wantedError: nil,
+		},
+		{
+			name: "Failed workflow",
+			workflow: &v1alpha1.Workflow{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-workflow",
+					Labels: map[string]string{
+						"repo":   "test-repo",
+						"commit": "test-commit",
+					},
+				},
+				Status: v1alpha1.WorkflowStatus{
+					Phase:   v1alpha1.WorkflowFailed,
+					Message: "something",
+				},
+			},
+			wantedError: nil,
+		},
+		{
+			name: "Error workflow",
+			workflow: &v1alpha1.Workflow{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-workflow",
+					Labels: map[string]string{
+						"repo":   "test-repo",
+						"commit": "test-commit",
+					},
+				},
+				Status: v1alpha1.WorkflowStatus{
+					Phase:   v1alpha1.WorkflowError,
+					Message: "something",
+				},
+			},
+			wantedError: nil,
+		},
+		{
+			name: "Pending workflow",
+			workflow: &v1alpha1.Workflow{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-workflow",
+					Labels: map[string]string{
+						"repo":   "test-repo",
+						"commit": "test-commit",
+					},
+				},
+				Status: v1alpha1.WorkflowStatus{
+					Phase:   v1alpha1.WorkflowPending,
+					Message: "something",
+				},
+			},
+			wantedError: nil,
+		},
+		{
+			name: "Running workflow",
+			workflow: &v1alpha1.Workflow{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-workflow",
+					Labels: map[string]string{
+						"repo":   "test-repo",
+						"commit": "test-commit",
+					},
+				},
+				Status: v1alpha1.WorkflowStatus{
+					Phase:   v1alpha1.WorkflowRunning,
+					Message: "something",
+				},
+			},
+			wantedError: nil,
+		},
+		{
+			name: "Missing label repo",
+			workflow: &v1alpha1.Workflow{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-workflow",
+					Labels: map[string]string{
+						"commit": "test-commit",
+					},
+				},
+				Status: v1alpha1.WorkflowStatus{
+					Phase:   v1alpha1.WorkflowSucceeded,
+					Message: "something",
+				},
+			},
+			wantedError: errors.New("some error"),
+		},
+		{
+			name: "Missing label commit",
+			workflow: &v1alpha1.Workflow{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-workflow",
+					Labels: map[string]string{
+						"repo": "test-repo",
+					},
+				},
+				Status: v1alpha1.WorkflowStatus{
+					Phase:   v1alpha1.WorkflowSucceeded,
+					Message: "something",
+				},
+			},
+			wantedError: errors.New("some error"),
+		},
+	}
+
+	// Create a mock configuration and clients
+	cfg := &conf.GlobalConfig{
+		WorkflowServerConfig: conf.WorkflowServerConfig{
+			ArgoAddress: "http://workflow-server",
+			Namespace:   "test-namespace",
+		},
+	}
+	clients := &clients.Clients{
+		GitProvider: &mockGitProvider{},
+	}
+
+	// Create a new githubNotifier instance
+	gn := NewGithubEventNotifier(cfg, clients)
+
+	// Call the Notify method
+
+	// Run test cases
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Call the function being tested
+			err := gn.Notify(&ctx, test.workflow)
+
+			// Use assert to check the equality of the error
+			if test.wantedError != nil {
+				assert.Error(err)
+				assert.NotNil(err)
+			} else {
+				assert.NoError(err)
+				assert.Nil(err)
+			}
+		})
+	}
+}
diff --git a/pkg/event_handler/main.go b/pkg/event_handler/main.go
@@ -0,0 +1,46 @@
+package event_handler
+
+import (
+	"context"
+	"github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
+	"github.com/rookout/piper/pkg/clients"
+	"github.com/rookout/piper/pkg/conf"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"log"
+)
+
+func Start(cfg *conf.GlobalConfig, clients *clients.Clients) {
+	ctx := context.Background() // TODO: use global context that initialized at main
+	labelSelector := &metav1.LabelSelector{
+		MatchExpressions: []metav1.LabelSelectorRequirement{
+			{Key: "piper.rookout.com/notified",
+				Operator: metav1.LabelSelectorOpExists},
+			{Key: "piper.rookout.com/notified",
+				Operator: metav1.LabelSelectorOpNotIn,
+				Values: []string{
+					string(v1alpha1.WorkflowSucceeded),
+					string(v1alpha1.WorkflowFailed),
+					string(v1alpha1.WorkflowError),
+				}}, // mean that there already completed and notified
+		},
+	}
+	watcher, err := clients.Workflows.Watch(&ctx, labelSelector)
+	if err != nil {
+		log.Printf("[event handler] Failed to watch workflow error:%s", err)
+		return
+	}
+
+	notifier := NewGithubEventNotifier(cfg, clients)
+	handler := &workflowEventHandler{
+		Clients:  clients,
+		Notifier: notifier,
+	}
+	go func() {
+		for event := range watcher.ResultChan() {
+			err = handler.Handle(ctx, &event)
+			if err != nil {
+				log.Printf("[event handler] failed to Handle workflow event %s", err) // ERROR
+			}
+		}
+	}()
+}
diff --git a/pkg/event_handler/types.go b/pkg/event_handler/types.go
@@ -0,0 +1,15 @@
+package event_handler
+
+import (
+	"github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
+	"golang.org/x/net/context"
+	"k8s.io/apimachinery/pkg/watch"
+)
+
+type EventHandler interface {
+	Handle(ctx context.Context, event *watch.Event) error
+}
+
+type EventNotifier interface {
+	Notify(ctx *context.Context, workflow *v1alpha1.Workflow) error
+}