fix(rest,satellite): unbreak CSI clone and restore, move the satellite stream to 3372

cmd/apiserver/main.go

@@ -198,7 +198,10 @@ func registerProbesAndREST(mgr manager.Manager, st store.Store, flags *apiserver
 		Store:     st,
 		Client:    mgr.GetClient(),
 		Namespace: namespace,
-		OnReady:   ready.MarkBound,
+		// The passphrase Secret is read past the cache, so no informer
+		// watches every Secret in the cluster for it.
+		SecretReader: mgr.GetAPIReader(),
+		OnReady:      ready.MarkBound,
 	})
 	if err != nil {
 		return errors.Wrap(err, "add rest server")

cmd/apiserver/secret_reader_wiring_test.go

@@ -0,0 +1,82 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"testing"
+)
+
+// Every rest.Server this binary builds reads the passphrase Secret through
+// the manager's API reader. Left unset, the server falls back to the cached
+// client, whose first Secret read starts a cluster-wide Secret informer that
+// the namespaced Role does not allow to list, and every passphrase read hangs.
+func TestRESTServerReadsSecretsPastTheCache(t *testing.T) {
+	t.Parallel()
+
+	fset := token.NewFileSet()
+
+	file, err := parser.ParseFile(fset, "main.go", nil, 0)
+	if err != nil {
+		t.Fatalf("parse main.go: %v", err)
+	}
+
+	servers := 0
+
+	ast.Inspect(file, func(n ast.Node) bool {
+		lit, ok := n.(*ast.CompositeLit)
+		if !ok || !isRESTServer(lit.Type) {
+			return true
+		}
+
+		servers++
+
+		if !setsSecretReaderToAPIReader(lit) {
+			t.Errorf("rest.Server at %s does not set SecretReader: mgr.GetAPIReader()", fset.Position(lit.Lbrace))
+		}
+
+		return true
+	})
+
+	if servers == 0 {
+		t.Fatal("main.go builds no rest.Server, so this guard checks nothing")
+	}
+}
+
+func isRESTServer(expr ast.Expr) bool {
+	sel, ok := expr.(*ast.SelectorExpr)
+	if !ok {
+		return false
+	}
+
+	pkg, ok := sel.X.(*ast.Ident)
+
+	return ok && pkg.Name == "rest" && sel.Sel.Name == "Server"
+}
+
+func setsSecretReaderToAPIReader(lit *ast.CompositeLit) bool {
+	for _, elt := range lit.Elts {
+		kv, ok := elt.(*ast.KeyValueExpr)
+		if !ok {
+			continue
+		}
+
+		key, ok := kv.Key.(*ast.Ident)
+		if !ok || key.Name != "SecretReader" {
+			continue
+		}
+
+		call, ok := kv.Value.(*ast.CallExpr)
+		if !ok {
+			return false
+		}
+
+		sel, ok := call.Fun.(*ast.SelectorExpr)
+
+		return ok && sel.Sel.Name == "GetAPIReader"
+	}
+
+	return false
+}

cmd/controller/main.go

@@ -314,7 +314,9 @@ func main() {
 			Store:     st,
 			Client:    mgr.GetClient(),
 			Namespace: controllerNamespace,
-			OnReady:   ready.MarkBound,
+			// As in cmd/apiserver: the passphrase Secret is read past the cache.
+			SecretReader: mgr.GetAPIReader(),
+			OnReady:      ready.MarkBound,
 		}); err != nil {
 			setupLog.Error(err, "Failed to register REST API server")
 			os.Exit(1)

cmd/controller/secret_reader_wiring_test.go

@@ -0,0 +1,82 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"testing"
+)
+
+// Every rest.Server this binary builds reads the passphrase Secret through
+// the manager's API reader. Left unset, the server falls back to the cached
+// client, whose first Secret read starts a cluster-wide Secret informer that
+// the namespaced Role does not allow to list, and every passphrase read hangs.
+func TestRESTServerReadsSecretsPastTheCache(t *testing.T) {
+	t.Parallel()
+
+	fset := token.NewFileSet()
+
+	file, err := parser.ParseFile(fset, "main.go", nil, 0)
+	if err != nil {
+		t.Fatalf("parse main.go: %v", err)
+	}
+
+	servers := 0
+
+	ast.Inspect(file, func(n ast.Node) bool {
+		lit, ok := n.(*ast.CompositeLit)
+		if !ok || !isRESTServer(lit.Type) {
+			return true
+		}
+
+		servers++
+
+		if !setsSecretReaderToAPIReader(lit) {
+			t.Errorf("rest.Server at %s does not set SecretReader: mgr.GetAPIReader()", fset.Position(lit.Lbrace))
+		}
+
+		return true
+	})
+
+	if servers == 0 {
+		t.Fatal("main.go builds no rest.Server, so this guard checks nothing")
+	}
+}
+
+func isRESTServer(expr ast.Expr) bool {
+	sel, ok := expr.(*ast.SelectorExpr)
+	if !ok {
+		return false
+	}
+
+	pkg, ok := sel.X.(*ast.Ident)
+
+	return ok && pkg.Name == "rest" && sel.Sel.Name == "Server"
+}
+
+func setsSecretReaderToAPIReader(lit *ast.CompositeLit) bool {
+	for _, elt := range lit.Elts {
+		kv, ok := elt.(*ast.KeyValueExpr)
+		if !ok {
+			continue
+		}
+
+		key, ok := kv.Key.(*ast.Ident)
+		if !ok || key.Name != "SecretReader" {
+			continue
+		}
+
+		call, ok := kv.Value.(*ast.CallExpr)
+		if !ok {
+			return false
+		}
+
+		sel, ok := call.Fun.(*ast.SelectorExpr)
+
+		return ok && sel.Sel.Name == "GetAPIReader"
+	}
+
+	return false
+}

cmd/satellite/coldstart_owned.go

@@ -0,0 +1,73 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"context"
+
+	"github.com/cockroachdb/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/rest"
+	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	blockstoriov1alpha1 "github.com/cozystack/blockstor/api/v1alpha1"
+	k8sstore "github.com/cozystack/blockstor/pkg/store/k8s"
+)
+
+// ownedKernelResources names the DRBD resources blockstor has a replica of on
+// nodeName: the definitions of the Resource objects scheduled there, read from
+// the API server rather than a cache, since the cache is not up yet.
+func ownedKernelResources(ctx context.Context, cfg *rest.Config, nodeName string) (map[string]struct{}, error) {
+	scheme := runtime.NewScheme()
+
+	err := blockstoriov1alpha1.AddToScheme(scheme)
+	if err != nil {
+		return nil, errors.Wrap(err, "register the blockstor types")
+	}
+
+	c, err := ctrlclient.New(cfg, ctrlclient.Options{Scheme: scheme})
+	if err != nil {
+		return nil, errors.Wrap(err, "build an API client")
+	}
+
+	return ownedOnNode(ctx, c, nodeName)
+}
+
+// ownedOnNode is ownedKernelResources over a given client.
+func ownedOnNode(ctx context.Context, c ctrlclient.Client, nodeName string) (map[string]struct{}, error) {
+	replicas, err := k8sstore.New(c).Resources().ListByNode(ctx, nodeName)
+	if err != nil {
+		return nil, errors.Wrapf(err, "list the replicas on %q", nodeName)
+	}
+
+	owned := make(map[string]struct{}, len(replicas))
+	for i := range replicas {
+		owned[replicas[i].Name] = struct{}{}
+	}
+
+	return owned, nil
+}
+
+// selectOrphans splits the kernel's resources that are not healthy into the
+// ones blockstor takes down, which it has a replica of here, and foreign ones
+// it leaves alone: another DRBD user's, or a resource whose Resource object is
+// already gone, which is not blockstor's to judge.
+func selectOrphans(names []string, healthy, owned map[string]struct{}) ([]string, []string) {
+	var orphans, foreign []string
+
+	for _, name := range names {
+		if _, ok := healthy[name]; ok {
+			continue
+		}
+
+		if _, ok := owned[name]; !ok {
+			foreign = append(foreign, name)
+
+			continue
+		}
+
+		orphans = append(orphans, name)
+	}
+
+	return orphans, foreign
+}