Add ability to pipe the agent to a custom integration for data mapping

Author: rocktavious

cmd/root.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"github.com/go-resty/resty/v2"
 	"os"
 	"runtime"
 	"strings"
@@ -51,7 +52,7 @@ var rootCmd = &cobra.Command{
 		var wg sync.WaitGroup
 		ctx := signal.Init(context.Background())
 		// go workers.NewWebhookWorker().Run(ctx, &wg)
-		go workers.NewK8SWorker(ctx, &wg, cluster, integration, configuration.Selectors, newClient(), resync, flush)
+		go workers.NewK8SWorker(ctx, &wg, cluster, integration, configuration.Selectors, newClient(), newRESTClient(), resync, flush)
 
 		time.Sleep(1 * time.Second)
 		wg.Wait()
@@ -162,6 +163,15 @@ func LoadConfig() (*config.Configuration, error) {
 	return &output, nil
 }
 
+func newRESTClient() *resty.Client {
+	client := opslevel.NewRestClient(
+		opslevel.SetURL(viper.GetString("api-url")),
+		opslevel.SetUserAgentExtra(fmt.Sprintf("agent-%s", _version)),
+		opslevel.SetTimeout(time.Second*time.Duration(viper.GetInt("api-timeout"))),
+	)
+	return client
+}
+
 func newClient() *opslevel.Client {
 	client := opslevel.NewGQLClient(
 		opslevel.SetAPIToken(viper.GetString("api-token")),

workers/k8s.go

@@ -3,6 +3,9 @@ package workers
 import (
 	"context"
 	"encoding/json"
+	"fmt"
+	"github.com/go-resty/resty/v2"
+	"strings"
 	"sync"
 	"time"
 
@@ -18,30 +21,39 @@ type K8SWorker struct {
 	cluster     string
 	integration string
 	client      *opslevel.Client
+	rest        *resty.Client
 	controller  *controller.Controller
 }
 
-func NewK8SWorker(ctx context.Context, wg *sync.WaitGroup, cluster string, integration string, selectors []controller.Selector, client *opslevel.Client, resync, flush time.Duration) {
+func NewK8SWorker(ctx context.Context, wg *sync.WaitGroup, cluster string, integration string, selectors []controller.Selector, client *opslevel.Client, rest *resty.Client, resync, flush time.Duration) {
 	controller.Run(ctx, wg, selectors, resync, flush, &K8SWorker{
-		client:      client,
 		cluster:     cluster,
 		integration: integration,
+		client:      client,
+		rest:        rest,
 	})
 }
 
 func (s *K8SWorker) Handle(evt controller.Event) {
 	kind := evt.ExternalKind()
 	id := evt.ExternalID(s.cluster)
 
-	switch evt.Op {
-	case controller.OpCreate, controller.OpUpdate:
-		value, err := s.parse(evt.New)
-		if err != nil {
-			log.Error().Err(err).Msgf("failed to convert k8s resource")
+	if strings.Contains(s.integration, "integrations/custom/webhook") {
+		switch evt.Op {
+		case controller.OpCreate, controller.OpUpdate:
+			s.sendEvent(kind, id, evt.New)
+		}
+	} else {
+		switch evt.Op {
+		case controller.OpCreate, controller.OpUpdate:
+			value, err := s.parse(evt.New)
+			if err != nil {
+				log.Error().Err(err).Msgf("failed to convert k8s resource")
+			}
+			s.sendUpsert(kind, id, value)
+		case controller.OpDelete:
+			s.sendDelete(kind, id)
 		}
-		s.sendUpsert(kind, id, value)
-	case controller.OpDelete:
-		s.sendDelete(kind, id)
 	}
 }
 
@@ -62,6 +74,26 @@ func (s *K8SWorker) parse(item *unstructured.Unstructured) (opslevel.JSON, error
 	return data, nil
 }
 
+func (s *K8SWorker) sendEvent(kind string, id string, value *unstructured.Unstructured) {
+	kind = strings.Replace(kind, "/", "_", -1)
+	if viper.GetBool("dry-run") {
+		log.Info().Msgf("[DRYRUN] POST %s | %s", kind, id)
+		log.Debug().Msgf("\t%#v", value)
+	} else {
+		url := fmt.Sprintf("%s?external_kind=%s", s.integration, kind)
+		resp, err := s.rest.R().SetBody(value).Post(url)
+		if err != nil {
+			log.Error().Err(err).Msgf("error during post")
+			return
+		}
+		if resp.StatusCode() > 299 {
+			log.Error().Msgf("%v", resp)
+			return
+		}
+		log.Info().Msgf("POST %s | %s", kind, id)
+	}
+}
+
 func (s *K8SWorker) sendUpsert(kind string, id string, value opslevel.JSON) {
 	var m struct {
 		Payload struct {