Merge pull request #609 from rancher/bugfixes

controllers/logging_dataprepper_controller.go

@@ -2,6 +2,7 @@ package controllers
 
 import (
 	"context"
+	"fmt"
 
 	loggingv1beta1 "github.com/rancher/opni/apis/logging/v1beta1"
 	"github.com/rancher/opni/pkg/resources"
@@ -16,7 +17,8 @@ import (
 
 type LoggingDataPrepperReconciler struct {
 	client.Client
-	scheme *runtime.Scheme
+	scheme      *runtime.Scheme
+	OpniCentral bool
 }
 
 // +kubebuilder:rbac:groups=logging.opni.io,resources=datapreppers,verbs=get;list;watch;create;update;patch;delete
@@ -30,7 +32,14 @@ func (r *LoggingDataPrepperReconciler) Reconcile(ctx context.Context, req ctrl.R
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
-	DataPrepperReconciler, err := dataprepper.NewReconciler(ctx, DataPrepper, r.Client)
+	var opts []dataprepper.ReconcilerOption
+
+	if r.OpniCentral {
+		opts = append(opts, dataprepper.WithForceInsecure())
+		opts = append(opts, dataprepper.WithURLOverride(fmt.Sprintf("https://opni-opensearch-svc.%s:9200", req.Namespace)))
+	}
+
+	DataPrepperReconciler, err := dataprepper.NewReconciler(ctx, DataPrepper, r.Client, opts...)
 	if err != nil {
 		return ctrl.Result{}, err
 	}

go.mod

@@ -21,7 +21,6 @@ require (
 	github.com/charmbracelet/bubbles v0.13.0
 	github.com/charmbracelet/bubbletea v0.22.1
 	github.com/charmbracelet/lipgloss v0.6.0
-	github.com/cockroachdb/errors v1.2.4
 	github.com/containerd/containerd v1.6.6
 	github.com/cortexproject/cortex v1.13.0
 	github.com/dbason/featureflags v0.0.0-20220803034705-b6242a8d72b2
@@ -33,7 +32,6 @@ require (
 	github.com/go-logr/zapr v1.2.0
 	github.com/go-playground/validator/v10 v10.11.0
 	github.com/go-test/deep v1.0.8
-	github.com/gogo/protobuf v1.3.2
 	github.com/gogo/status v1.1.0
 	github.com/golang/mock v1.6.1-0.20220512030613-73266f9366fc
 	github.com/golang/snappy v0.0.4
@@ -60,7 +58,7 @@ require (
 	github.com/kralicky/kmatch v0.0.0-20220713045459-85a252b9275e
 	github.com/kralicky/ragu v1.0.0-rc1
 	github.com/kralicky/spellbook v0.0.0-20220829172922-3d415e02ee8a
-	github.com/kralicky/totem v1.1.0-rc1
+	github.com/kralicky/totem v1.1.0-rc2
 	github.com/kralicky/yaml/v3 v3.0.0-20220520012407-b0e7050bd81d
 	github.com/kubernetes-sigs/node-feature-discovery-operator v0.2.1-0.20210826163723-568b36491208
 	github.com/lestrrat-go/backoff/v2 v2.0.8
@@ -267,6 +265,7 @@ require (
 	github.com/gocql/gocql v0.0.0-20200526081602-cd04bd7f22a7 // indirect
 	github.com/gofrs/uuid v4.2.0+incompatible // indirect
 	github.com/gogo/googleapis v1.4.0 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang-jwt/jwt/v4 v4.4.1 // indirect
 	github.com/golang-migrate/migrate/v4 v4.7.0 // indirect
 	github.com/golang/glog v1.0.0 // indirect

go.sum

@@ -1886,8 +1886,8 @@ github.com/kralicky/ragu v1.0.0-rc1 h1:CFFDEA1Tt8AL+olth4vV/IZ5J0q0TWin204K3qS1V
 github.com/kralicky/ragu v1.0.0-rc1/go.mod h1:82qibIMOw+3FUmKU01DKIkFXm7Vcoit6Y3P/+wjhgFI=
 github.com/kralicky/spellbook v0.0.0-20220829172922-3d415e02ee8a h1:RfpqOGzhMSF0OWvI8AqYxlvhuVZ72EQiYu9ClM89bKg=
 github.com/kralicky/spellbook v0.0.0-20220829172922-3d415e02ee8a/go.mod h1:qp6R4XKeuJk2j8nHke4Y9rBEpc2HkCjXR/gssL3OpOU=
-github.com/kralicky/totem v1.1.0-rc1 h1:3brmz73ZFGy6S1Q/qyCbHxqgegRDHPKLTvoNlx3bPwQ=
-github.com/kralicky/totem v1.1.0-rc1/go.mod h1:p8nlcJV48ZiUUtWpwWzVltuY6V9ZaRBCbzZNGN7FJpA=
+github.com/kralicky/totem v1.1.0-rc2 h1:1HD6BEyTfphqjuyTEGdmgTzMBdOZCtZkUylVk6FXrjg=
+github.com/kralicky/totem v1.1.0-rc2/go.mod h1:p8nlcJV48ZiUUtWpwWzVltuY6V9ZaRBCbzZNGN7FJpA=
 github.com/kralicky/yaml/v3 v3.0.0-20220520012407-b0e7050bd81d h1:kLfaaFdmCHKZCvL4DzQ7T9YsAVSBqZez34zaegldkls=
 github.com/kralicky/yaml/v3 v3.0.0-20220520012407-b0e7050bd81d/go.mod h1:z4cKsjE0B5RlhFSbWcZNh70UEjtwrj8fpG1QXyU2KuI=
 github.com/kralicky/zap v1.21.1-0.20220517214003-da0ae127f4db h1:OhMQACnkI4y5uEciBkaVGzPFpS/5LkPT1uHnOv84Fcw=

packages/opni-agent/opni-agent/charts/templates/rbac_v2.yaml

@@ -15,6 +15,8 @@ rules:
   - namespaces
   verbs:
   - get
+  - list
+  - watch
 - apiGroups:
   - ""
   resources:
@@ -47,6 +49,13 @@ rules:
   - statefulsets
   - statefulsets/finalizers
   - statefulsets/status
+  - replicasets
+  verbs:
+  - "*"
+- apiGroups:
+  - ""
+  resources:
+  - pods
   verbs:
   - "*"
 - apiGroups:
@@ -63,6 +72,16 @@ rules:
   - persistentvolumeclaims
   verbs:
   - "*"
+- apiGroups:
+  - policy
+  resources:
+  - podsecuritypolicies
+  verbs:
+  - create
+  - use
+  - get
+  - list
+  - watch
 
 # prometheus resources
 - apiGroups:

packages/opni-agent/opni-agent/charts/templates/statefulset_v2.yaml

@@ -106,6 +106,9 @@ spec:
           args:
           - --health-probe-bind-address=:7081
           - --metrics-bind-address=127.0.0.1:7080
+          {{- if .Values.bootstrapInCluster.enabled }}
+          - --central
+          {{- end }}
           securityContext:
             {{- toYaml .Values.securityContext | nindent 12 }}
           image: "{{ template "system_default_registry" . }}{{ .Values.image.repository }}:{{ .Values.image.tag | default $versionTag }}"

packages/opni-agent/opni-agent/charts/templates/webhooks.yaml

@@ -1,6 +1,6 @@
 {{- $namespace := printf "%s" (include "opni-agent.namespace" .) }}
 {{- if not .Values.bootstrapInCluster.enabled -}}
-{{- if (or (eq .Values.agent.version "v2") .Values.logging.enabled ) }}
+{{- if .Values.logging.enabled }}
 apiVersion: admissionregistration.k8s.io/v1
 kind: MutatingWebhookConfiguration
 metadata:

pkg/bootstrap/client.go

@@ -126,6 +126,8 @@ func (c *ClientConfig) bootstrapJoin(ctx context.Context) (*bootstrapv1.Bootstra
 	}
 	cc, err := grpc.DialContext(ctx, c.Endpoint,
 		append(c.DialOpts,
+			grpc.WithBlock(),
+			grpc.FailOnNonTempDialError(true),
 			grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)),
 			grpc.WithChainStreamInterceptor(otelgrpc.StreamClientInterceptor()),
 			grpc.WithChainUnaryInterceptor(otelgrpc.UnaryClientInterceptor()),
@@ -139,7 +141,7 @@ func (c *ClientConfig) bootstrapJoin(ctx context.Context) (*bootstrapv1.Bootstra
 	client := bootstrapv1.NewBootstrapClient(cc)
 
 	var peer peer.Peer
-	resp, err := client.Join(ctx, &bootstrapv1.BootstrapJoinRequest{}, grpc.Peer(&peer))
+	resp, err := client.Join(ctx, &bootstrapv1.BootstrapJoinRequest{}, grpc.Peer(&peer), grpc.WaitForReady(true))
 	if err != nil {
 		return nil, nil, fmt.Errorf("join request failed: %w", err)
 	}

pkg/bootstrap/client_v2.go

@@ -124,6 +124,8 @@ func (c *ClientConfigV2) bootstrapJoin(ctx context.Context) (*bootstrapv2.Bootst
 	}
 	cc, err := grpc.DialContext(ctx, c.Endpoint,
 		append(c.DialOpts,
+			grpc.WithBlock(),
+			grpc.FailOnNonTempDialError(true),
 			grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)),
 			grpc.WithChainStreamInterceptor(otelgrpc.StreamClientInterceptor()),
 			grpc.WithChainUnaryInterceptor(otelgrpc.UnaryClientInterceptor()),
@@ -137,7 +139,7 @@ func (c *ClientConfigV2) bootstrapJoin(ctx context.Context) (*bootstrapv2.Bootst
 	client := bootstrapv2.NewBootstrapClient(cc)
 
 	var peer peer.Peer
-	resp, err := client.Join(ctx, &bootstrapv2.BootstrapJoinRequest{}, grpc.Peer(&peer))
+	resp, err := client.Join(ctx, &bootstrapv2.BootstrapJoinRequest{}, grpc.Peer(&peer), grpc.WaitForReady(true))
 	if err != nil {
 		return nil, nil, fmt.Errorf("join request failed: %w", err)
 	}

pkg/gateway/grpc.go

@@ -3,6 +3,7 @@ package gateway
 import (
 	"context"
 	"net"
+	"sync"
 	"time"
 
 	"github.com/samber/lo"
@@ -51,7 +52,8 @@ type GatewayGRPCServer struct {
 	logger     *zap.SugaredLogger
 	serverOpts []grpc.ServerOption
 
-	services []util.ServicePack[any]
+	servicesMu sync.Mutex
+	services   []util.ServicePack[any]
 }
 
 func NewGRPCServer(
@@ -85,9 +87,11 @@ func (s *GatewayGRPCServer) ListenAndServe(ctx context.Context) error {
 		grpc.MaxRecvMsgSize(32*1024*1024), // 32MB
 	)...)
 	healthv1.RegisterHealthServer(server, health.NewServer())
+	s.servicesMu.Lock()
 	for _, services := range s.services {
 		server.RegisterService(services.Unpack())
 	}
+	s.servicesMu.Unlock()
 
 	s.logger.With(
 		"address", listener.Addr().String(),
@@ -106,5 +110,7 @@ func (s *GatewayGRPCServer) ListenAndServe(ctx context.Context) error {
 }
 
 func (s *GatewayGRPCServer) RegisterService(desc *grpc.ServiceDesc, impl any) {
+	s.servicesMu.Lock()
+	defer s.servicesMu.Unlock()
 	s.services = append(s.services, util.PackService(desc, impl))
 }

pkg/opni/commands/client.go

@@ -36,6 +36,7 @@ func BuildClientCmd() *cobra.Command {
 		disableUsage bool
 		echoVersion  bool
 		logLevel     string
+		opniCentral  bool
 	)
 
 	run := func(signalCtx context.Context) error {
@@ -90,7 +91,9 @@ func BuildClientCmd() *cobra.Command {
 			return err
 		}
 
-		if err = (&controllers.LoggingDataPrepperReconciler{}).SetupWithManager(mgr); err != nil {
+		if err = (&controllers.LoggingDataPrepperReconciler{
+			OpniCentral: opniCentral,
+		}).SetupWithManager(mgr); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "Logging DataPrepper")
 			return err
 		}
@@ -149,6 +152,7 @@ func BuildClientCmd() *cobra.Command {
 	cmd.Flags().StringVar(&logLevel, "log-level", "info", "log level (debug, info, warning, error)")
 	cmd.Flags().StringVar(&metricsAddr, "metrics-bind-address", ":7080", "The address the metric endpoint binds to.")
 	cmd.Flags().StringVar(&probeAddr, "health-probe-bind-address", ":7081", "The address the probe endpoint binds to.")
+	cmd.Flags().BoolVarP(&opniCentral, "central", "c", false, "run controllers in Opni central cluster mode")
 	cmd.Flags().BoolVarP(&echoVersion, "version", "v", false, "print the version and exit")
 	features.DefaultMutableFeatureGate.AddFlag(cmd.Flags())