adaptation: close plugin if initial sync fails.

pkg/adaptation/adaptation.go

@@ -593,10 +593,10 @@ func (r *Adaptation) acceptPluginConnections(l net.Listener) error {
 			}
 
 			r.requestPluginSync()
-
 			err = r.syncFn(ctx, p.synchronize)
 			if err != nil {
-				log.Infof(ctx, "failed to synchronize plugin: %v", err)
+				log.Errorf(ctx, "failed to synchronize plugin: %v", err)
+				p.close()
 			} else {
 				r.Lock()
 				r.plugins = append(r.plugins, p)
@@ -607,7 +607,6 @@ func (r *Adaptation) acceptPluginConnections(l net.Listener) error {
 				r.Unlock()
 				log.Infof(ctx, "plugin %q connected and synchronized", p.name())
 			}
-
 			r.finishedPluginSync()
 		}
 	}()

pkg/adaptation/adaptation_suite_test.go

@@ -257,6 +257,29 @@ var _ = Describe("Plugin connection", func() {
 		Expect(protoEqual(plugin.ctrs["ctr1"], runtime.ctrs["ctr1"])).Should(BeTrue(),
 			protoDiff(plugin.ctrs["ctr1"], runtime.ctrs["ctr1"]))
 	})
+
+	It("close plugins on failed synchronization", func() {
+		var (
+			runtime = s.runtime
+			plugin0 = s.plugins[0]
+			plugin1 = &mockPlugin{idx: "10", name: "bar"}
+			timeout = time.After(startupTimeout)
+		)
+
+		s.Startup()
+
+		Expect(plugin0.Events()).Should(
+			ConsistOf(
+				PluginConfigured,
+				PluginSynchronized,
+			),
+		)
+
+		runtime.failSync = true
+
+		s.StartPlugins(plugin1)
+		Expect(plugin1.Wait(PluginDisconnected, timeout)).To(Succeed())
+	})
 })
 
 var _ = Describe("Pod and container requests and events", func() {

pkg/adaptation/suite_test.go

@@ -158,6 +158,7 @@ type mockRuntime struct {
 	ctrs    map[string]*api.Container
 
 	updateFn nri.UpdateFn
+	failSync bool
 }
 
 func (m *mockRuntime) Start(dir string) error {
@@ -228,6 +229,11 @@ func (m *mockRuntime) synchronize(ctx context.Context, cb nri.SyncCB) error {
 	}
 
 	_, err := cb(ctx, pods, ctrs)
+
+	if m.failSync {
+		return fmt.Errorf("mock runtime forced synchronize failure")
+	}
+
 	return err
 }