feat: workflow status listener - RK-19190

.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  \

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() {
@@ -51,10 +50,6 @@ func main() {
 		panic(err)
 	}
 
-	//err = common.Git.UnsetWebhook()
-	//if err != nil {
-	//	panic(err)
-	//}
-
+	event_handler.Start(cfg, globalClients)
 	server.Start(cfg, globalClients)
 }

docs/CONTRIBUTING.md

@@ -68,12 +68,15 @@ make helm
 To make it easy to develop locally, please run the following
 
 Prerequisites :
-1. install ngrok
+1. install helm
+2. install kubectl
+3. isntall docker
+4. install ngrok
 ```bash
 brew install ngrok
 ```
-2. install docker
-3. install kind
+5. install docker
+6. install kind
 ```bash
 brew install kind
 ```

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
+}

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() 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)
+			}
+		})
+	}
+}

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
+			}
+		}
+	}()
+}

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
+}