support webhooks

README.md

@@ -119,6 +119,7 @@ app:
     webhook:
         url: ${BEETLE_WEBHOOK_URL:- }
         retry: ${BEETLE_WEBHOOK_RETRY:-3}
+        token: ${BEETLE_WEBHOOK_TOKEN:- }
 
 # Log configs
 log:

config.dist.yml

@@ -65,6 +65,7 @@ app:
     webhook:
         url: ${BEETLE_WEBHOOK_URL:- }
         retry: ${BEETLE_WEBHOOK_RETRY:-3}
+        token: ${BEETLE_WEBHOOK_TOKEN:- }
 
 # Log configs
 log:

config.testing.yml

@@ -65,6 +65,7 @@ app:
     webhook:
         url: ${BEETLE_WEBHOOK_URL:- }
         retry: ${BEETLE_WEBHOOK_RETRY:-3}
+        token: ${BEETLE_WEBHOOK_TOKEN:- }
 
 # Log configs
 log:

internal/app/controller/daemon.go

@@ -5,13 +5,18 @@
 package controller
 
 import (
+	"context"
+	"fmt"
+	"net/http"
+	"strconv"
 	"time"
 
 	"github.com/clivern/beetle/internal/app/model"
 	"github.com/clivern/beetle/internal/app/module"
 
 	"github.com/prometheus/client_golang/prometheus"
 	log "github.com/sirupsen/logrus"
+	"github.com/spf13/viper"
 )
 
 var (
@@ -49,9 +54,20 @@ func Daemon() {
 	var pendingJobsCount int
 	var failedJobsCount int
 	var successfulJobsCount int
+	var job model.Job
+	var parentJob model.Job
+	var deploymentRequest model.DeploymentRequest
+	var payload string
 
+	httpClient := module.NewHTTPClient()
 	db := module.Database{}
 
+	retry, err := strconv.Atoi(viper.GetString("app.webhook.retry"))
+
+	if err != nil {
+		panic(err.Error())
+	}
+
 	for {
 		err = db.AutoConnect()
 
@@ -80,6 +96,60 @@ func Daemon() {
 		successJobs.Set(float64(successfulJobsCount))
 
 		// Run Pending Jobs (HTTP Notification)
+		job = db.GetPendingJobByType(model.JobDeploymentNotify)
+
+		if job.ID > 0 {
+			if job.Retry > retry {
+				now := time.Now()
+				job.Status = model.JobFailed
+				job.RunAt = &now
+				job.Result = fmt.Sprintf("Failed to deliver the notification")
+				db.UpdateJobByID(&job)
+			} else {
+				deploymentRequest.LoadFromJSON([]byte(job.Payload))
+
+				if job.Parent > 0 {
+					parentJob = db.GetJobByID(job.Parent)
+
+					if parentJob.ID > 0 {
+						deploymentRequest.Status = parentJob.Status
+					}
+				}
+
+				payload, _ = deploymentRequest.ConvertToJSON()
+
+				response, err := httpClient.Post(
+					context.TODO(),
+					viper.GetString("app.webhook.url"),
+					payload,
+					map[string]string{},
+					map[string]string{
+						"Content-Type":      "application/json",
+						"X-AUTH-TOKEN":      viper.GetString("app.webhook.token"),
+						"X-NOTIFICATION-ID": job.UUID,
+						"X-DEPLOYMENT-ID":   parentJob.UUID,
+					},
+				)
+
+				if httpClient.GetStatusCode(response) != http.StatusOK || err != nil {
+					job.Status = model.JobFailed
+					job.Result = fmt.Sprintf("Failed to deliver the notification")
+				} else {
+					job.Status = model.JobSuccess
+					job.Result = fmt.Sprintf("Notification delivered successfully")
+				}
+
+				if job.Status == model.JobFailed && job.Retry <= retry {
+					job.Status = model.JobPending
+				}
+
+				now := time.Now()
+				job.Retry++
+				job.RunAt = &now
+				db.UpdateJobByID(&job)
+			}
+		}
+
 		time.Sleep(2 * time.Second)
 	}
 }

internal/app/controller/worker.go

@@ -84,6 +84,7 @@ func Worker(workerID int, messages <-chan string) {
 			job.RunAt = &now
 			job.Result = fmt.Sprintf("Invalid job payload, UUID %s", messageObj.UUID)
 			db.UpdateJobByID(&job)
+			db.ReleaseChildJobs(job.ID)
 			continue
 		}
 
@@ -100,7 +101,7 @@ func Worker(workerID int, messages <-chan string) {
 		}).Info(`Worker accepted deployment request`)
 
 		// Notify if there is a webhook
-		if viper.GetString("webhook.url") != "" {
+		if viper.GetString("app.webhook.url") != "" {
 			uuid = util.GenerateUUID4()
 
 			for db.JobExistByUUID(uuid) {
@@ -110,7 +111,7 @@ func Worker(workerID int, messages <-chan string) {
 			db.CreateJob(&model.Job{
 				UUID:    uuid,
 				Payload: job.Payload,
-				Status:  model.JobPending,
+				Status:  model.JobOnHold,
 				Parent:  messageObj.Job,
 				Type:    model.JobDeploymentNotify,
 			})
@@ -125,7 +126,7 @@ func Worker(workerID int, messages <-chan string) {
 				"request_application": deploymentRequest.Application,
 				"request_version":     deploymentRequest.Version,
 				"request_strategy":    deploymentRequest.Strategy,
-				"webhook_url":         viper.GetString("webhook.url"),
+				"webhook_url":         viper.GetString("app.webhook.url"),
 			}).Info(`HTTP webhook enabled`)
 		} else {
 			log.WithFields(log.Fields{
@@ -161,6 +162,7 @@ func Worker(workerID int, messages <-chan string) {
 			job.RunAt = &now
 			job.Result = fmt.Sprintf("Worker can not find the cluster, UUID %s", messageObj.UUID)
 			db.UpdateJobByID(&job)
+			db.ReleaseChildJobs(job.ID)
 			continue
 		}
 
@@ -184,6 +186,7 @@ func Worker(workerID int, messages <-chan string) {
 			job.RunAt = &now
 			job.Result = fmt.Sprintf("Worker unable to ping cluster, UUID %s", messageObj.UUID)
 			db.UpdateJobByID(&job)
+			db.ReleaseChildJobs(job.ID)
 			continue
 		}
 
@@ -207,6 +210,7 @@ func Worker(workerID int, messages <-chan string) {
 			job.RunAt = &now
 			job.Result = fmt.Sprintf("Failure during deployment, UUID %s", messageObj.UUID)
 			db.UpdateJobByID(&job)
+			db.ReleaseChildJobs(job.ID)
 			continue
 		}
 
@@ -226,5 +230,6 @@ func Worker(workerID int, messages <-chan string) {
 		job.RunAt = &now
 		job.Result = "Deployment finished successfully"
 		db.UpdateJobByID(&job)
+		db.ReleaseChildJobs(job.ID)
 	}
 }

internal/app/model/job.go

@@ -19,6 +19,9 @@ var (
 	// JobSuccess success job type
 	JobSuccess = "SUCCESS"
 
+	// JobOnHold on hold job type
+	JobOnHold = "ON_HOLD"
+
 	// JobDeploymentUpdate deployment update
 	JobDeploymentUpdate = "deployment.update"
 

internal/app/model/request.go

@@ -28,6 +28,7 @@ type DeploymentRequest struct {
 	Application string `json:"application"`
 	Version     string `json:"version"`
 	Strategy    string `json:"strategy"`
+	Status      string `json:"status"`
 }
 
 // LoadFromJSON update object from json

internal/app/module/database.go

@@ -168,6 +168,15 @@ func (db *Database) GetJobByUUID(uuid string) model.Job {
 	return job
 }
 
+// GetPendingJobByType gets a job by uuid
+func (db *Database) GetPendingJobByType(jobType string) model.Job {
+	job := model.Job{}
+
+	db.Connection.Where("status = ? AND type = ?", model.JobPending, jobType).First(&job)
+
+	return job
+}
+
 // CountJobs count jobs by status
 func (db *Database) CountJobs(status string) int {
 	count := 0
@@ -196,3 +205,12 @@ func (db *Database) UpdateJobByID(job *model.Job) {
 func (db *Database) Close() error {
 	return db.Connection.Close()
 }
+
+// ReleaseChildJobs count jobs by status
+func (db *Database) ReleaseChildJobs(parentID int) {
+	db.Connection.Model(&model.Job{}).Where(
+		"parent = ? AND status = ?",
+		parentID,
+		model.JobOnHold,
+	).Update("status", model.JobPending)
+}