Integration Tests: Adds injector

cmd/injector/app/app.go

@@ -97,6 +97,7 @@ func Run() {
 	}
 
 	inj, err := service.NewInjector(service.Options{
+		Port:                    opts.Port,
 		AuthUIDs:                uids,
 		Config:                  cfg,
 		DaprClient:              daprClient,

cmd/injector/options/options.go

@@ -26,6 +26,7 @@ import (
 type Options struct {
 	HealthzPort int
 	Kubeconfig  string
+	Port        int
 	Logger      logger.Options
 	Metrics     *metrics.Options
 }
@@ -53,6 +54,7 @@ func New(origArgs []string) *Options {
 	fs.SortFlags = true
 
 	fs.IntVar(&opts.HealthzPort, "healthz-port", 8080, "The port used for health checks")
+	fs.IntVar(&opts.Port, "port", 4000, "The port used for the injector service")
 
 	if home := homedir.HomeDir(); home != "" {
 		fs.StringVar(&opts.Kubeconfig, "kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file")

pkg/injector/service/injector.go

@@ -38,7 +38,6 @@ import (
 )
 
 const (
-	port                                      = 4000
 	getKubernetesServiceAccountTimeoutSeconds = 10
 	systemGroup                               = "system:masters"
 	serviceAccountUserInfoPrefix              = "system:serviceaccount:"
@@ -72,6 +71,7 @@ type Options struct {
 	Config     Config
 	DaprClient scheme.Interface
 	KubeClient kubernetes.Interface
+	Port       int
 
 	ControlPlaneNamespace   string
 	ControlPlaneTrustDomain string
@@ -140,7 +140,7 @@ func NewInjector(opts Options) (Injector, error) {
 			runtime.NewScheme(),
 		).UniversalDeserializer(),
 		server: &http.Server{
-			Addr:              fmt.Sprintf(":%d", port),
+			Addr:              fmt.Sprintf(":%d", opts.Port),
 			Handler:           mux,
 			ReadHeaderTimeout: 10 * time.Second,
 		},

tests/integration/framework/binary/binary.go

@@ -31,7 +31,7 @@ import (
 func BuildAll(t *testing.T) {
 	t.Helper()
 
-	binaryNames := []string{"daprd", "placement", "sentry", "operator"}
+	binaryNames := []string{"daprd", "placement", "sentry", "operator", "injector"}
 
 	var wg sync.WaitGroup
 	wg.Add(len(binaryNames))

tests/integration/framework/process/injector/injector.go

@@ -0,0 +1,155 @@
+/*
+Copyright 2024 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package injector
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/spiffe/go-spiffe/v2/spiffeid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	admissionregistrationv1 "k8s.io/api/admissionregistration/v1"
+
+	"github.com/dapr/dapr/tests/integration/framework/binary"
+	"github.com/dapr/dapr/tests/integration/framework/process"
+	"github.com/dapr/dapr/tests/integration/framework/process/exec"
+	"github.com/dapr/dapr/tests/integration/framework/process/kubernetes"
+	"github.com/dapr/dapr/tests/integration/framework/util"
+)
+
+type Injector struct {
+	kubeapi  *kubernetes.Kubernetes
+	exec     process.Interface
+	freeport *util.FreePort
+
+	port        int
+	namespace   string
+	metricsPort int
+	healthzPort int
+}
+
+func New(t *testing.T, fopts ...Option) *Injector {
+	t.Helper()
+
+	fp := util.ReservePorts(t, 3)
+	opts := options{
+		logLevel:      "info",
+		enableMetrics: true,
+		port:          fp.Port(t, 0),
+		metricsPort:   fp.Port(t, 1),
+		healthzPort:   fp.Port(t, 2),
+		sidecarImage:  "integration.dapr.io/dapr",
+	}
+
+	for _, fopt := range fopts {
+		fopt(&opts)
+	}
+
+	require.NotNil(t, opts.sentry, "sentry is required")
+	require.NotNil(t, opts.namespace, "namespace is required")
+
+	mobj, err := json.Marshal(new(admissionregistrationv1.MutatingWebhookConfiguration))
+	require.NoError(t, err)
+	kubeapi := kubernetes.New(t, kubernetes.WithBaseOperatorAPI(t,
+		spiffeid.RequireTrustDomainFromString(opts.sentry.TrustDomain(t)),
+		*opts.namespace,
+		opts.sentry.Port(),
+	),
+		kubernetes.WithPath("/apis/admissionregistration.k8s.io/v1/mutatingwebhookconfigurations/dapr-sidecar-injector", func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Add("Content-Type", "application/json")
+			w.Write(mobj)
+		}),
+	)
+
+	args := []string{
+		"-log-level=" + opts.logLevel,
+		"-port=" + strconv.Itoa(opts.port),
+		"-metrics-port=" + strconv.Itoa(opts.metricsPort),
+		"-healthz-port=" + strconv.Itoa(opts.healthzPort),
+		"-kubeconfig=" + kubeapi.KubeconfigPath(t),
+	}
+
+	return &Injector{
+		kubeapi: kubeapi,
+		exec: exec.New(t,
+			binary.EnvValue("injector"), args,
+			append(
+				opts.execOpts,
+				exec.WithEnvVars(t,
+					"KUBERNETES_SERVICE_HOST", "anything",
+					"NAMESPACE", *opts.namespace,
+					"SIDECAR_IMAGE", opts.sidecarImage,
+					"DAPR_TRUST_ANCHORS_FILE", opts.sentry.TrustAnchorsFile(t),
+					"DAPR_CONTROL_PLANE_TRUST_DOMAIN", opts.sentry.TrustDomain(t),
+					"DAPR_SENTRY_ADDRESS", opts.sentry.Address(),
+				),
+			)...,
+		),
+		freeport:    fp,
+		port:        opts.port,
+		metricsPort: opts.metricsPort,
+		healthzPort: opts.healthzPort,
+		namespace:   *opts.namespace,
+	}
+}
+
+func (i *Injector) Run(t *testing.T, ctx context.Context) {
+	i.kubeapi.Run(t, ctx)
+	i.freeport.Free(t)
+	i.exec.Run(t, ctx)
+}
+
+func (i *Injector) Cleanup(t *testing.T) {
+	i.exec.Cleanup(t)
+	i.kubeapi.Cleanup(t)
+}
+
+func (i *Injector) WaitUntilRunning(t *testing.T, ctx context.Context) {
+	client := util.HTTPClient(t)
+	//nolint:testifylint
+	assert.EventuallyWithT(t, func(t *assert.CollectT) {
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("http://localhost:%d/healthz", i.healthzPort), nil)
+		if !assert.NoError(t, err) {
+			return
+		}
+		resp, err := client.Do(req)
+		if !assert.NoError(t, err) {
+			return
+		}
+		assert.NoError(t, resp.Body.Close())
+		assert.Equal(t, http.StatusOK, resp.StatusCode)
+	}, time.Second*5, 10*time.Millisecond)
+}
+
+func (i *Injector) Port() int {
+	return i.port
+}
+
+func (i *Injector) Address() string {
+	return "localhost:" + strconv.Itoa(i.port)
+}
+
+func (i *Injector) MetricsPort() int {
+	return i.metricsPort
+}
+
+func (i *Injector) HealthzPort() int {
+	return i.healthzPort
+}

tests/integration/framework/process/injector/options.go

@@ -0,0 +1,92 @@
+/*
+Copyright 2024 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package injector
+
+import (
+	"github.com/dapr/dapr/tests/integration/framework/process/exec"
+	"github.com/dapr/dapr/tests/integration/framework/process/sentry"
+)
+
+// options contains the options for running Injector in integration tests.
+type options struct {
+	execOpts []exec.Option
+
+	logLevel      string
+	namespace     *string
+	port          int
+	enableMetrics bool
+	metricsPort   int
+	healthzPort   int
+
+	sidecarImage string
+
+	sentry *sentry.Sentry
+}
+
+// Option is a function that configures the process.
+type Option func(*options)
+
+func WithExecOptions(execOptions ...exec.Option) Option {
+	return func(o *options) {
+		o.execOpts = execOptions
+	}
+}
+
+func WithLogLevel(level string) Option {
+	return func(o *options) {
+		o.logLevel = level
+	}
+}
+
+func WithNamespace(namespace string) Option {
+	return func(o *options) {
+		o.namespace = &namespace
+	}
+}
+
+func WithPort(port int) Option {
+	return func(o *options) {
+		o.port = port
+	}
+}
+
+func WithMetricsPort(port int) Option {
+	return func(o *options) {
+		o.metricsPort = port
+	}
+}
+
+func WithEnableMetrics(enable bool) Option {
+	return func(o *options) {
+		o.enableMetrics = enable
+	}
+}
+
+func WithHealthzPort(port int) Option {
+	return func(o *options) {
+		o.healthzPort = port
+	}
+}
+
+func WithSidecarImage(image string) Option {
+	return func(o *options) {
+		o.sidecarImage = image
+	}
+}
+
+func WithSentry(sentry *sentry.Sentry) Option {
+	return func(o *options) {
+		o.sentry = sentry
+	}
+}

tests/integration/framework/process/kubernetes/options.go

@@ -86,6 +86,10 @@ func WithClusterStatefulSetList(t *testing.T, ss *appsv1.StatefulSetList) Option
 	return handleClusterListResource(t, "/apis/apps/v1/statefulsets", ss)
 }
 
+func WithClusterServiceAccountList(t *testing.T, services *corev1.ServiceAccountList) Option {
+	return handleClusterListResource(t, "/api/v1/serviceaccounts", services)
+}
+
 func WithDaprConfigurationGet(t *testing.T, config *configapi.Configuration) Option {
 	return handleGetResource(t, "/apis/dapr.io/v1alpha1", "configurations", config.Namespace, config.Name, config)
 }
@@ -118,6 +122,7 @@ func WithBaseOperatorAPI(t *testing.T, td spiffeid.TrustDomain, ns string, sentr
 			WithClusterServiceList(t, &corev1.ServiceList{TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "ServiceList"}}),
 			WithClusterStatefulSetList(t, &appsv1.StatefulSetList{TypeMeta: metav1.TypeMeta{APIVersion: "apps/v1", Kind: "StatefulSetList"}}),
 			WithClusterDeploymentList(t, &appsv1.DeploymentList{TypeMeta: metav1.TypeMeta{APIVersion: "apps/v1", Kind: "DeploymentList"}}),
+			WithClusterServiceAccountList(t, &corev1.ServiceAccountList{TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "ServiceAccountList"}}),
 			WithClusterDaprComponentList(t, &compapi.ComponentList{TypeMeta: metav1.TypeMeta{APIVersion: "dapr.io/v1alpha1", Kind: "ComponentList"}}),
 			WithClusterDaprHTTPEndpointList(t, &httpendapi.HTTPEndpointList{TypeMeta: metav1.TypeMeta{APIVersion: "dapr.io/v1alpha1", Kind: "HTTPEndpointList"}}),
 			WithClusterDaprResiliencyList(t, &resapi.ResiliencyList{TypeMeta: metav1.TypeMeta{APIVersion: "dapr.io/v1alpha1", Kind: "ResiliencyList"}}),

tests/integration/framework/process/operator/options.go

@@ -17,7 +17,7 @@ import (
 	"github.com/dapr/dapr/tests/integration/framework/process/exec"
 )
 
-// options contains the options for running Sentry in integration tests.
+// options contains the options for running Operator in integration tests.
 type options struct {
 	execOpts []exec.Option
 

tests/integration/framework/process/sentry/sentry.go

@@ -49,6 +49,7 @@ type Sentry struct {
 	port        int
 	healthzPort int
 	metricsPort int
+	trustDomain *string
 }
 
 func New(t *testing.T, fopts ...Option) *Sentry {
@@ -136,6 +137,7 @@ func New(t *testing.T, fopts ...Option) *Sentry {
 		port:        opts.port,
 		metricsPort: opts.metricsPort,
 		healthzPort: opts.healthzPort,
+		trustDomain: opts.trustDomain,
 	}
 }
 
@@ -190,6 +192,11 @@ func (s *Sentry) HealthzPort() int {
 	return s.healthzPort
 }
 
+func (s *Sentry) TrustDomain(t *testing.T) string {
+	require.NotNil(t, s.trustDomain)
+	return *s.trustDomain
+}
+
 // DialGRPC dials the sentry using the given context and returns a grpc client
 // connection.
 func (s *Sentry) DialGRPC(t *testing.T, ctx context.Context, sentryID string) *grpc.ClientConn {