Resume publish when creating a new publisher (#226)

* Resume publish when creating a new publisher

* Make labels consistent

controllers/unleash_controller.go

@@ -32,7 +32,12 @@ import (
 	"github.com/nais/unleasherator/pkg/unleashclient"
 )
 
-const unleashFinalizer = "unleash.nais.io/finalizer"
+const (
+	unleashFinalizer                  = "unleash.nais.io/finalizer"
+	unleashPublishMetricStatusSending = "sending"
+	unleashPublishMetricStatusSuccess = "success"
+	unleashPublishMetricStatusFailed  = "failed"
+)
 
 var (
 	// unleashStatus is a Prometheus metric which will be used to expose the status of the Unleash instances
@@ -321,21 +326,23 @@ func (r *UnleashReconciler) publish(ctx context.Context, unleash *unleashv1.Unle
 	}
 
 	log.Info("Publishing Unleash instance to federation")
+	// Count the number of Unleash instances published
+	unleashPublished.WithLabelValues("provisioned", unleashPublishMetricStatusSending).Inc()
 
 	token, err := unleash.AdminToken(ctx, r.Client, r.OperatorNamespace)
 	if err != nil {
-		unleashPublished.WithLabelValues("provisioned", "failed").Inc()
+		unleashPublished.WithLabelValues("provisioned", unleashPublishMetricStatusFailed).Inc()
 		log.Error(err, "Failed to fetch API token")
 		return fmt.Errorf("publish could not fetch API token: %w", err)
 	}
 
 	err = r.Federation.Publisher.Publish(ctx, unleash, string(token))
 	if err != nil {
-		unleashPublished.WithLabelValues("provisioned", "failed").Inc()
+		unleashPublished.WithLabelValues("provisioned", unleashPublishMetricStatusFailed).Inc()
 		return fmt.Errorf("publish could not publish Unleash instance: %w", err)
 	}
 
-	unleashPublished.WithLabelValues("provisioned", "success").Inc()
+	unleashPublished.WithLabelValues("provisioned", unleashPublishMetricStatusSuccess).Inc()
 	return nil
 }
 

controllers/unleash_controller_test.go

@@ -161,7 +161,11 @@ var _ = Describe("Unleash controller", func() {
 			Expect(mockPublisher.AssertCalled(GinkgoT(), "Publish", mock.AnythingOfType("*context.valueCtx"), mock.AnythingOfType("*unleash_nais_io_v1.Unleash"), mock.AnythingOfType("string"))).To(BeTrue())
 			Expect(mockPublisher.AssertNumberOfCalls(GinkgoT(), "Publish", 1)).To(BeTrue())
 
-			val, err := promCounterVecVal(unleashPublished, "provisioned", "success")
+			val, err := promCounterVecVal(unleashPublished, "provisioned", unleashPublishMetricStatusSending)
+			Expect(err).To(BeNil())
+			Expect(val).To(Equal(float64(1)))
+
+			val, err = promCounterVecVal(unleashPublished, "provisioned", unleashPublishMetricStatusSuccess)
 			Expect(err).To(BeNil())
 			Expect(val).To(Equal(float64(1)))
 

pkg/federation/publisher.go

@@ -59,5 +59,8 @@ func (p *publisher) Publish(ctx context.Context, unleash *unleashv1.Unleash, api
 }
 
 func NewPublisher(client *pubsub.Client, topic *pubsub.Topic) Publisher {
+	// Fix for the following pubsub error, this clears the ordering key for the topic when the publisher is created
+	// pubsub: Publishing for ordering key, order, paused due to previous error. Call topic.ResumePublish(orderingKey) before resuming publishing
+	topic.ResumePublish(pubsubOrderingKey)
 	return &publisher{client: client, topic: topic}
 }