changes from review and fix configmap label

cmd/mtping/main.go

@@ -31,5 +31,5 @@ func main() {
 	ctx := signals.NewContext()
 	ctx = adapter.WithConfigMapWatcherEnabled(ctx)
 	ctx = adapter.WithInjectorEnabled(ctx)
-	adapter.MainWithContext(ctx, "pingsource-mt-adapter", mtping.NewEnvConfig, mtping.NewAdapter)
+	adapter.MainWithContext(ctx, component, mtping.NewEnvConfig, mtping.NewAdapter)
 }

pkg/adapter/mtping/adapter.go

@@ -52,7 +52,9 @@ func NewAdapter(ctx context.Context, _ adapter.EnvConfigAccessor, ceClient cloud
 // Start implements adapter.Adapter
 func (a *mtpingAdapter) Start(ctx context.Context) error {
 	a.logger.Info("Starting job runner...")
-	a.runner.Start(ctx.Done())
+	if err := a.runner.Start(ctx.Done()); err != nil {
+		return err
+	}
 
 	a.logger.Infof("runner stopped")
 	return nil

pkg/adapter/mtping/runner.go

@@ -138,6 +138,7 @@ func message(body string) interface{} {
 }
 
 func (a *cronJobsRunner) updateFromConfigMap(cm *corev1.ConfigMap) {
+	a.Logger.Info("synchronizing configmap")
 	data, ok := cm.Data[cache.ResourcesKey]
 	if !ok {
 		// Shouldn't happened.
@@ -165,14 +166,17 @@ func (a *cronJobsRunner) updateFromConfigMap(cm *corev1.ConfigMap) {
 		// Is the schedule already cached?
 		if cfgid, ok := a.entryids[key]; ok {
 			if !equality.Semantic.DeepEqual(cfgid.config, cfg) {
+				a.Logger.Infof("updating schedule", zap.Any("key", key))
 				// Recreate cronjob
 				a.RemoveSchedule(cfgid.entryID)
 				cfgid.entryID = a.AddSchedule(cfg)
 				cfgid.config = &cfg
+
 			} else {
 				// cron jon exists and correctly configure. noop.
 			}
 		} else {
+			a.Logger.Infof("adding schedule", zap.Any("key", key))
 			// Create cronjob
 			a.entryids[key] = entryIdConfig{
 				entryID: a.AddSchedule(cfg),
@@ -185,6 +189,7 @@ func (a *cronJobsRunner) updateFromConfigMap(cm *corev1.ConfigMap) {
 
 	for key := range keys {
 		if cfgid, ok := a.entryids[key]; ok {
+			a.Logger.Infof("deleting schedule", zap.Any("key", key))
 			a.RemoveSchedule(cfgid.entryID)
 		}
 	}

pkg/reconciler/pingsource/pingsource.go

@@ -64,7 +64,6 @@ const (
 	component                = "pingsource"
 	mtcomponent              = "pingsource-mt-adapter"
 	mtadapterName            = "pingsource-mt-adapter"
-	mtconfigmapName          = "config-pingsource-mt-adapter"
 	stadapterClusterRoleName = "knative-eventing-pingsource-adapter"
 )
 

pkg/utils/cache/persisted_store.go

@@ -206,7 +206,7 @@ func (p *persistedStore) load() (*corev1.ConfigMap, error) {
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      p.name,
 				Namespace: p.namespace,
-				Annotations: map[string]string{
+				Labels: map[string]string{
 					ComponentLabelKey: p.component,
 				},
 			},

pkg/utils/cache/persisted_store_test.go

@@ -58,12 +58,12 @@ func TestPersistedStore(t *testing.T) {
 		cmName = "test-cm-name"
 	)
 	cs := fake.NewSimpleClientset()
-	created := make(chan struct{})
+	created := make(chan runtime.Object)
 	updated := make(chan runtime.Object)
 	done := make(chan bool)
 	cs.PrependReactor("create", "configmaps",
 		func(action ktesting.Action) (bool, runtime.Object, error) {
-			close(created)
+			created <- action.(ktesting.CreateAction).GetObject()
 			return false, nil, nil
 		},
 	)
@@ -93,8 +93,12 @@ func TestPersistedStore(t *testing.T) {
 	informer.Add(kr)
 
 	select {
-	case <-created:
+	case obj := <-created:
 		// We expect the configmap to be created.
+		cm := obj.(*corev1.ConfigMap)
+		if value, ok := cm.Labels[ComponentLabelKey]; !ok || value != "my-component" {
+			t.Fatalf("Missing %s label. Got %v", ComponentLabelKey, cm)
+		}
 	case <-time.After(1 * time.Second):
 		t.Fatal("Timed out waiting for configmap creation.")
 	}
@@ -158,12 +162,12 @@ func TestPersistedStoreInterrupted(t *testing.T) {
 		cmName = "test-cm-name"
 	)
 	cs := fake.NewSimpleClientset()
-	created := make(chan struct{})
+	created := make(chan runtime.Object)
 	updated := make(chan runtime.Object)
 	cs.PrependReactor("create", "configmaps",
 		func(action ktesting.Action) (bool, runtime.Object, error) {
 			time.Sleep(1 * time.Second)
-			close(created)
+			created <- action.(ktesting.CreateAction).GetObject()
 			return false, nil, nil
 		},
 	)
@@ -196,8 +200,12 @@ func TestPersistedStoreInterrupted(t *testing.T) {
 	informer.Add(kr2) // interrupt
 
 	select {
-	case <-created:
+	case obj := <-created:
 		// We expect the configmap to be created.
+		cm := obj.(*corev1.ConfigMap)
+		if value, ok := cm.Labels[ComponentLabelKey]; !ok || value != "my-component" {
+			t.Fatalf("Missing %s label. Got %v", ComponentLabelKey, cm)
+		}
 	case <-time.After(2 * time.Second):
 		t.Fatal("Timed out waiting for configmap creation.")
 	}