feat(gcp): add step to send events to GCP (#4896)

* add gcp token handling

* add initial step

* publish events

* add test cases

* fix test case

---------

Co-authored-by: Jordi van Liempt <[email protected]>

cmd/gcpPublishEvent.go

@@ -0,0 +1,45 @@
+package cmd
+
+import (
+	"github.com/SAP/jenkins-library/pkg/gcp"
+	"github.com/SAP/jenkins-library/pkg/log"
+	"github.com/SAP/jenkins-library/pkg/telemetry"
+	"github.com/pkg/errors"
+)
+
+func gcpPublishEvent(config gcpPublishEventOptions, telemetryData *telemetry.CustomData) {
+	err := runGcpPublishEvent(&config, telemetryData)
+	if err != nil {
+		log.Entry().WithError(err).Fatal("step execution failed")
+	}
+}
+
+func runGcpPublishEvent(config *gcpPublishEventOptions, telemetryData *telemetry.CustomData) error {
+	// cdevents.NewCDEvent("")
+	// create event
+	// pipelineID := ""
+	// pipelineURL := ""
+	// pipelineName := ""
+	// pipelineSource := ""
+
+	// create event data
+	data := []byte{}
+	// data, err := events.CreatePipelineRunStartedCDEventAsBytes(pipelineID, pipelineName, pipelineSource, pipelineURL)
+	// if err != nil {
+	// 	return errors.Wrap(err, "failed to create event data")
+	// }
+
+	// get federated token
+	token, err := gcp.GetFederatedToken(config.GcpProjectNumber, config.GcpWorkloadIDentityPool, config.GcpWorkloadIDentityPoolProvider, config.OIDCToken)
+	if err != nil {
+		return errors.Wrap(err, "failed to get federated token")
+	}
+
+	// publish event
+	err = gcp.Publish(config.GcpProjectNumber, config.Topic, token, data)
+	if err != nil {
+		return errors.Wrap(err, "failed to publish event")
+	}
+
+	return nil
+}

cmd/gcpPublishEvent_generated_test.go

@@ -0,0 +1,20 @@
+//go:build unit
+// +build unit
+
+package cmd
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGcpPublishEventCommand(t *testing.T) {
+	t.Parallel()
+
+	testCmd := GcpPublishEventCommand()
+
+	// only high level testing performed - details are tested in step generation procedure
+	assert.Equal(t, "gcpPublishEvent", testCmd.Use, "command name incorrect")
+
+}

cmd/piper.go

@@ -92,6 +92,7 @@ var GeneralConfig GeneralConfigOptions
 func Execute() {
 	log.Entry().Infof("Version %s", GitCommit)
 
+	rootCmd.AddCommand(GcpPublishEventCommand())
 	rootCmd.AddCommand(ArtifactPrepareVersionCommand())
 	rootCmd.AddCommand(ConfigCommand())
 	rootCmd.AddCommand(DefaultsCommand())

pkg/gcp/event.go

@@ -0,0 +1,62 @@
+package gcp
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+
+	"github.com/pkg/errors"
+)
+
+const api_url = "https://pubsub.googleapis.com/v1/projects/%s/topics/%s:publish"
+
+// https://pkg.go.dev/cloud.google.com/go/pubsub#Message
+type EventMessage struct {
+	Data []byte `json:"data"`
+}
+
+type Event struct {
+	Messages []EventMessage `json:"messages"`
+}
+
+func Publish(projectNumber string, topic string, token string, data []byte) error {
+	ctx := context.Background()
+
+	// build event
+	event := Event{
+		Messages: []EventMessage{{
+			Data: data,
+		}},
+	}
+
+	// marshal event
+	eventBytes, err := json.Marshal(event)
+	if err != nil {
+		return errors.Wrap(err, "failed to marshal event")
+	}
+
+	// create request
+	request, err := http.NewRequestWithContext(ctx, http.MethodPost, fmt.Sprintf(api_url, projectNumber, topic), bytes.NewReader(eventBytes))
+	if err != nil {
+		return errors.Wrap(err, "failed to create request")
+	}
+
+	// add headers
+	request.Header.Set("Content-Type", "application/json")
+	request.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token))
+
+	// send request
+	response, err := http.DefaultClient.Do(request)
+	if err != nil {
+		return errors.Wrap(err, "failed to send request")
+	}
+	if response.StatusCode != http.StatusOK {
+		return fmt.Errorf("invalid status code: %v", response.StatusCode)
+	}
+
+	//TODO: read response & messageIds
+
+	return nil
+}

pkg/gcp/event_test.go

@@ -0,0 +1,45 @@
+package gcp
+
+import (
+	"fmt"
+	"net/http"
+	"testing"
+
+	"github.com/jarcoal/httpmock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+func TestPublish(t *testing.T) {
+	t.Run("success", func(t *testing.T) {
+		// init
+		projectNumber := "PROJECT_NUMBER"
+		topic := "TOPIC"
+		token := "TOKEN"
+		data := []byte(mock.Anything)
+
+		apiurl := fmt.Sprintf(api_url, projectNumber, topic)
+
+		mockResponse := map[string]interface{}{
+			"messageIds": []string{"10721501285371497"},
+		}
+
+		// mock
+		httpmock.Activate()
+		defer httpmock.DeactivateAndReset()
+		httpmock.RegisterResponder(http.MethodPost, apiurl,
+			func(req *http.Request) (*http.Response, error) {
+				assert.Contains(t, req.Header, "Authorization")
+				assert.Equal(t, req.Header.Get("Authorization"), "Bearer TOKEN")
+				assert.Contains(t, req.Header, "Content-Type")
+				assert.Equal(t, req.Header.Get("Content-Type"), "application/json")
+				return httpmock.NewJsonResponse(http.StatusOK, mockResponse)
+			},
+		)
+
+		// test
+		err := Publish(projectNumber, topic, token, data)
+		// asserts
+		assert.NoError(t, err)
+	})
+}