Kafka ACL changes for firehose

cli/serve.go

@@ -132,12 +132,13 @@ func (l moduleConfigLookup) ModuleConfigs(ctx context.Context, moduleURN string)
 func setupRegistry(store *postgres.Store) module.Registry {
 	supported := []module.Descriptor{
 		kubernetes.Module,
-		firehose.Module,
 		job.Module,
 		kafka.Module,
 		flink.Module,
-		// dagger resolves ACL kafka streams by fetching their resource internally
-		// (no dependency), so it needs a resource getter backed by the store.
+		// dagger and firehose resolve ACL kafka streams by fetching their
+		// resource internally (no dependency), so they need a resource getter
+		// backed by the store.
+		firehose.Module(store.GetByURN),
 		dagger.Module(store.GetByURN),
 	}
 

docs/modules/firehose.md

@@ -34,4 +34,72 @@ type moduleConfig struct {
 | `ChartVersion` | `string` Chart version you want to use. |
 | `Firehose` | `struct` Holds firehose configuration. |
 
-Detailed JSONSchema for config can be referenced [here](https://github.com/goto/entropy/blob/main/modules/firehose/schema/config.json).
+Detailed JSONSchema for config can be referenced [here](https://github.com/goto/entropy/blob/main/modules/firehose/schema/config.json).
+
+## Kafka ACL (SASL/SSL) source streams
+
+A firehose names its kafka resource through `stream_name`, and flags a secured one with
+`stream_security_enabled`. The stream's security profile — the same `security` block the
+kafka module exposes on its output — is then resolved during Plan, in this order:
+
+1. an inline `stream_security` entry (prefetched by Dex),
+2. a declared kafka dependency keyed by the stream name,
+3. the kafka resource fetched internally by URN
+   (`orn:entropy:kafka:<project>:<stream_name>`), with no dependency declared.
+
+The flag exists so plaintext firehoses cost no lookup: the caller already knows whether a
+profile exists, because Dex reads the same resource for the broker address. `stream_name`
+is sent for plaintext streams too, so a stream that *loses* its ACLs gets the wiring a
+previous plan left behind cleared.
+
+The wiring mirrors odin's firehose adapter, so a migrated firehose renders the same pod
+spec it does today. From the resolved profile the module injects the
+`SOURCE_KAFKA_CONSUMER_CONFIG_*` env variables — security protocol, SASL mechanism, SSL
+protocol, truststore type, and for a stream carrying certs a fixed
+`SSL_TRUSTSTORE_LOCATION` of `/etc/secret/kafka-cert/truststore.p12` (`.jks` for JKS) plus
+the `SSL_TRUSTSTORE_FILENAME` the chart uses to select that key out of the secret. The cert
+gets its own subdirectory rather than odin's bare `/etc/secret`, because the chart already
+mounts its sink-credential secret there. OAUTHBEARER
+additionally gets the `OAuthBearerLoginModule` JAAS string and the pod login callback
+handler class.
+
+No secret value ever passes through entropy. The material is described as references in the
+`acl` config, rendered as the `kafka_security` chart value alongside `service_account`:
+
+| field | what the chart does with it |
+| :--- | :--- |
+| `ssl_config_credential` + `truststore_filename` | mounts that existing secret at `/etc/secret/kafka-cert`, selecting the filename as both key and path |
+| `truststore_password` (`secretName` + `key`) | renders `SOURCE_KAFKA_CONSUMER_CONFIG_SSL_TRUSTSTORE_PASSWORD` as a `secretKeyRef` env var |
+| `jaas_config_credential` | mounts the PLAIN/SCRAM `jaas.conf` secret at `/etc/secret/kafka` |
+| `kafka_token_enabled` | adds the projected service-account token (`audience: kafka`) at `/var/run/secrets/kafka/serviceaccount` |
+
+PLAIN/SCRAM streams never inline credentials or a JAAS string: they read a mounted
+`jaas.conf`, and the module appends
+`-Djava.security.auth.login.config=/etc/secret/kafka/jaas.conf` to `_JAVA_OPTIONS` (the
+rest of that variable is left alone). The secret is the profile's `acls[team].secretName`
+when set, otherwise odin's `<team>-<stream>-jaas` convention.
+
+Plaintext firehoses are untouched — no injected config, no `kafka_security` value, no
+chart value changes.
+
+Naming a stream also relaxes the schema's `SOURCE_KAFKA_BROKERS` requirement: when the
+stream is resolved, brokers are filled from its `url` unless the payload set them. Plan
+fails if neither supplied them, rather than deploying a firehose with no brokers — note
+that a stream named without the flag resolves nothing, so brokers must be explicit. Dex
+sends the address itself in every case, so its value always wins.
+
+The callback handler class and default service account are deployment level settings under
+the module's driver config:
+
+```json
+{
+  "kafka_security": {
+    "sasl_login_callback_handler_class": "io.gtflabs.kafka.security.oauthbearer.kubernetes.PodLoginCallbackHandler",
+    "service_account": "aegis-kafka"
+  }
+}
+```
+
+> Requires firehose chart `0.2.1` or later (`goto/charts`, `stable/firehose`). Chart `0.2.0`
+> renders none of this — it has no `serviceAccountName`, projected volume or `secretKeyRef`
+> support — so an ACL firehose on it starts without its secrets and fails to authenticate.

modules/firehose/config.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/goto/entropy/core/resource"
 	"github.com/goto/entropy/modules"
+	kafkamod "github.com/goto/entropy/modules/kafka"
 	"github.com/goto/entropy/pkg/errors"
 	"github.com/goto/entropy/pkg/validator"
 )
@@ -67,6 +68,37 @@ type Config struct {
 	ChartValues   *ChartValues  `json:"chart_values,omitempty"`
 	InitContainer InitContainer `json:"init_container,omitempty"`
 	Autoscaler    *Autoscaler   `json:"autoscaler,omitempty"`
+
+	// Team owns the firehose. It selects the credential reference from a
+	// stream's PLAIN/SCRAM ACL list.
+	Team string `json:"team,omitempty"`
+
+	// StreamName is the name of the kafka resource backing SOURCE_KAFKA_BROKERS.
+	// It is the key used to resolve the stream's security profile. Setting it
+	// also makes this module the owner of the SASL/SSL env variables, which are
+	// rebuilt from the stream on every plan.
+	StreamName string `json:"stream_name,omitempty"`
+
+	// StreamSecurityEnabled says the named stream carries a security profile, so
+	// the driver resolves its kafka resource internally (by URN, without a
+	// declared dependency). The caller already knows this — Dex reads the same
+	// resource for the broker address — so a plaintext firehose costs no lookup.
+	StreamSecurityEnabled bool `json:"stream_security_enabled,omitempty"`
+
+	// StreamSecurity holds the kafka security profile fetched by Dex, keyed by
+	// stream name. References only — never inline secret values.
+	StreamSecurity map[string]*kafkamod.SecurityProfile `json:"stream_security,omitempty"`
+
+	// ACL describes the secret material an ACL (SASL_SSL/OAUTHBEARER, SSL,
+	// PLAIN/SCRAM) source stream needs mounted. It is computed by
+	// applyStreamSecurity from the resolved stream security profile and holds
+	// references only — never secret values.
+	ACL *ACLConfig `json:"acl,omitempty"`
+
+	// ServiceAccount, when set, becomes the pod's service account. It is the
+	// OAuth identity authorized for ACL streams. Empty preserves the chart's
+	// default service account.
+	ServiceAccount string `json:"service_account,omitempty"`
 }
 
 type Telegraf struct {

modules/firehose/driver.go

@@ -90,8 +90,13 @@ type firehoseDriver struct {
 	kubeGetPod        kubeGetPodFn
 	kubeGetDeployment kubeGetDeploymentFn
 	consumerReset     consumerResetFn
+	getResource       ResourceGetter
 }
 
+// ResourceGetter fetches a resource by URN. It lets the driver resolve an ACL
+// kafka stream internally, without declaring it as a dependency.
+type ResourceGetter func(ctx context.Context, urn string) (*resource.Resource, error)
+
 type (
 	kubeDeployFn        func(ctx context.Context, isCreate bool, conf kube.Config, hc helm.ReleaseConfig) error
 	kubeGetPodFn        func(ctx context.Context, conf kube.Config, ns string, labels map[string]string) ([]kube.Pod, error)
@@ -147,6 +152,10 @@ type driverConf struct {
 	KubeDeployTimeout int `json:"kube_deploy_timeout_seconds"`
 
 	Autoscaler FirehoseAutoscaler `json:"autoscaler,omitempty"`
+
+	// KafkaSecurity holds the deployment level settings used when wiring an ACL
+	// (SASL/SSL) source stream.
+	KafkaSecurity KafkaSecurity `json:"kafka_security,omitempty"`
 }
 
 type FirehoseAutoscaler struct {
@@ -396,6 +405,35 @@ func (fd *firehoseDriver) getHelmRelease(res resource.Resource, conf Config,
 		"mountSecrets": mountSecrets,
 	}
 
+	// ACL (SASL/SSL) source support, mirroring odin's firehose manifest: the
+	// chart mounts the referenced secrets and renders the truststore password as
+	// a secretKeyRef. Both keys are omitted for plaintext firehoses, which keeps
+	// their rendered chart values unchanged.
+	if conf.ACL != nil {
+		aclValues := map[string]any{}
+		if conf.ACL.SSLConfigCredential != "" {
+			aclValues["ssl_config_credential"] = conf.ACL.SSLConfigCredential
+			aclValues["truststore_filename"] = conf.ACL.TruststoreFilename
+		}
+		if conf.ACL.TruststorePassword != nil {
+			aclValues["truststore_password"] = map[string]any{
+				"secretName": conf.ACL.TruststorePassword.SecretName,
+				"key":        conf.ACL.TruststorePassword.Key,
+			}
+		}
+		if conf.ACL.JaasConfigCredential != "" {
+			aclValues["jaas_config_credential"] = conf.ACL.JaasConfigCredential
+		}
+		if conf.ACL.KafkaTokenEnabled {
+			aclValues["kafka_token_enabled"] = true
+		}
+		rc.Values["kafka_security"] = aclValues
+	}
+
+	if conf.ServiceAccount != "" {
+		rc.Values["service_account"] = conf.ServiceAccount
+	}
+
 	if conf.Autoscaler != nil {
 		rc.Values["autoscaler"], err = conf.Autoscaler.GetHelmValues(conf)
 		if err != nil {

modules/firehose/driver_plan.go

@@ -21,23 +21,23 @@ var suffixRegex = regexp.MustCompile(`^([A-Za-z0-9-]+)-([0-9]+)$`)
 
 var errCauseInvalidNamespaceUpdate = "cannot update kube namespace of a running firehose"
 
-func (fd *firehoseDriver) Plan(_ context.Context, exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) {
+func (fd *firehoseDriver) Plan(ctx context.Context, exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) {
 	switch act.Name {
 	case module.CreateAction:
-		return fd.planCreate(exr, act)
+		return fd.planCreate(ctx, exr, act)
 
 	case ResetAction:
-		return fd.planReset(exr, act)
+		return fd.planReset(ctx, exr, act)
 
 	case ResetV2Action:
-		return fd.planResetV2(exr, act)
+		return fd.planResetV2(ctx, exr, act)
 
 	default:
-		return fd.planChange(exr, act)
+		return fd.planChange(ctx, exr, act)
 	}
 }
 
-func (fd *firehoseDriver) planChange(exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) {
+func (fd *firehoseDriver) planChange(ctx context.Context, exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) {
 	curConf, err := readConfig(exr.Resource, exr.Resource.Spec.Configs, fd.conf)
 	if err != nil {
 		return nil, err
@@ -84,6 +84,12 @@ func (fd *firehoseDriver) planChange(exr module.ExpandedResource, act module.Act
 			}
 		}
 
+		// resolve the ACL source stream (SASL/SSL consumer config + pod mounts).
+		// No-op for plaintext sources.
+		if err := fd.applyStreamSecurity(ctx, exr, newConf); err != nil {
+			return nil, errors.ErrInvalid.WithMsgf("failed to resolve source stream").WithCausef("%s", err.Error())
+		}
+
 		curConf = newConf
 
 	case ScaleAction:
@@ -144,12 +150,18 @@ func (fd *firehoseDriver) planChange(exr module.ExpandedResource, act module.Act
 	return &exr.Resource, nil
 }
 
-func (fd *firehoseDriver) planCreate(exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) {
+func (fd *firehoseDriver) planCreate(ctx context.Context, exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) {
 	conf, err := readConfig(exr.Resource, act.Params, fd.conf)
 	if err != nil {
 		return nil, err
 	}
 
+	// resolve the ACL source stream (SASL/SSL consumer config + pod mounts).
+	// No-op for plaintext sources.
+	if err := fd.applyStreamSecurity(ctx, exr, conf); err != nil {
+		return nil, errors.ErrInvalid.WithMsgf("failed to resolve source stream").WithCausef("%s", err.Error())
+	}
+
 	chartVals, err := mergeChartValues(&fd.conf.ChartValues, conf.ChartValues)
 	if err != nil {
 		return nil, err
@@ -193,7 +205,7 @@ func (fd *firehoseDriver) planCreate(exr module.ExpandedResource, act module.Act
 	return &exr.Resource, nil
 }
 
-func (fd *firehoseDriver) planResetV2(exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) {
+func (fd *firehoseDriver) planResetV2(ctx context.Context, exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) {
 	resetValue, err := kafka.ParseResetV2Params(act.Params)
 	if err != nil {
 		return nil, err
@@ -206,6 +218,10 @@ func (fd *firehoseDriver) planResetV2(exr module.ExpandedResource, act module.Ac
 		return nil, err
 	}
 
+	if err := fd.applyStreamSecurity(ctx, exr, curConf); err != nil {
+		return nil, errors.ErrInvalid.WithMsgf("failed to resolve source stream").WithCausef("%s", err.Error())
+	}
+
 	curConf.ResetOffset = resetValue
 
 	exr.Resource.Spec.Configs = modules.MustJSON(curConf)
@@ -225,7 +241,7 @@ func (fd *firehoseDriver) planResetV2(exr module.ExpandedResource, act module.Ac
 	return &exr.Resource, nil
 }
 
-func (fd *firehoseDriver) planReset(exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) {
+func (fd *firehoseDriver) planReset(ctx context.Context, exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) {
 	resetValue, err := kafka.ParseResetParams(act.Params)
 	if err != nil {
 		return nil, err
@@ -238,6 +254,10 @@ func (fd *firehoseDriver) planReset(exr module.ExpandedResource, act module.Acti
 		return nil, err
 	}
 
+	if err := fd.applyStreamSecurity(ctx, exr, curConf); err != nil {
+		return nil, errors.ErrInvalid.WithMsgf("failed to resolve source stream").WithCausef("%s", err.Error())
+	}
+
 	curConf.ResetOffset = resetValue
 	curConf.EnvVariables[SourceKafkaConsumerAutoOffsetReset] = resetValue
 	curConf.EnvVariables[confKeyConsumerID], err = getNewConsumerGroupID(curConf.EnvVariables[confKeyConsumerID])