Conversation
knative-prow
Bot
added
the
kind/enhancement
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: aliok The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
knative-prow
Bot
added
the
size/XXL
There was a problem hiding this comment.
🟡 Changes recommended
There are correctness and UX gaps around KEDA trigger configurations (unsupported trigger types/combos) and TriggerAuthentication construction that can lead to broken Kafka scaling at runtime.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
Adds KEDA-based Kafka consumer-lag autoscaling support and restructures func.yaml autoscaling configuration so KEDA/KPA settings live under a new top-level scale field, with a migration to preserve backwards compatibility.
Changes:
- Introduces top-level
scalewithkeda/kpasub-keys, updates validation, schema, docs, and a migration fromdeploy.options.scale. - Adds KEDA Kafka scaling implementation that creates a
ScaledObject+ optionalTriggerAuthentication, and updates the KEDA remover to clean these up. - Updates deployers, CLI, and tests to use
Deploy.ActiveDeployer/Deploy.ActiveExposeand the newFunction.Scalelocation.
File summaries
| File | Description |
|---|---|
| schema/func_yaml-schema.json | Updates schema for scale top-level move and new KEDA/KPA structures. |
| pkg/pipelines/tekton/pipelines_provider.go | Renames deployed-state fields to ActiveDeployer/ActiveExpose. |
| pkg/mock/deployer.go | Updates mock deployer to use Deploy.ActiveDeployer. |
| pkg/knative/deployer.go | Reads autoscaling annotations from Function.Scale (KPA sub-key). |
| pkg/keda/remover.go | Adds explicit deletion of Kafka KEDA scaling resources during remove. |
| pkg/keda/kafka_scaling.go | Implements unstructured KEDA ScaledObject/TriggerAuthentication helpers. |
| pkg/keda/kafka_scaling_test.go | Unit tests for Kafka scaling object construction helpers. |
| pkg/keda/kafka_scaling_int_test.go | Integration test verifying KEDA resources created/removed in-cluster. |
| pkg/keda/deployer.go | Wires trigger selection and creates Kafka scaling resources during deploy. |
| pkg/k8s/wait.go | Treats Deployments with desired replicas 0 as “available” for waits. |
| pkg/k8s/deployer.go | Seeds Deployment replicas from Function.Scale.Min. |
| pkg/functions/function.go | Adds Function.Scale, renames deployed-state fields, hooks new validation. |
| pkg/functions/function_scale.go | New deployer-aware scale validation entry point. |
| pkg/functions/function_options.go | Reworks scale types into KEDA/KPA; keeps legacy options-scale for migration only. |
| pkg/functions/function_options_unit_test.go | Moves scale tests to ValidateScale; retains resource validation tests. |
| pkg/functions/function_migrations.go | Adds 0.37.0 migration to move scale to top-level + KPA flattening. |
| pkg/functions/function_migrations_unit_test.go | Adds unit tests covering the new scale migration scenarios. |
| pkg/functions/client.go | Uses Deploy.ActiveDeployer/ActiveExpose for deploy/remove state tracking. |
| pkg/functions/client_test.go | Updates tests for renamed deployed-state fields. |
| pkg/deployer/testing/integration_test_helper.go | Updates integration helper to set Function.Scale and ActiveExpose. |
| pkg/config/config.go | Uses Deploy.ActiveDeployer when seeding global config. |
| e2e/e2e_recorder_test.go | Updates E2E recorder test to set Function.Scale. |
| e2e/e2e_expose_test.go | Updates E2E exposure tests for trigger requirement and ActiveExpose/Deployer. |
| docs/reference/func_yaml.md | Documents new scale and run.kafka structures and examples. |
| docs/reference/func_deploy.md | Updates deployer flag help to reflect broader KEDA trigger support. |
| cmd/func-util/main.go | Uses Deploy.ActiveDeployer as fallback for effective deployer. |
| cmd/deploy.go | Updates help text and deployed-state field usage; adjusts exposure-record warning logic. |
| cmd/deploy_test.go | Updates deploy tests for ActiveDeployer/Expose and KEDA trigger requirement. |
| cmd/delete_test.go | Updates delete tests for ActiveDeployer and KEDA trigger requirement. |
Review details
Suppressed comments (1)
docs/reference/func_yaml.md:218
- The KEDA example configures both an HTTP trigger and a Kafka trigger, but the current deployer implementation uses an HTTPScaledObject for HTTP and a separate ScaledObject for Kafka; those generally can’t both target the same Deployment. Update the example to a supported single-trigger configuration (or add an explicit limitation note).
- type: http
targetValue: 200
- type: kafka
lagThreshold: 5
activationLagThreshold: 0
- Files reviewed: 29/29 changed files
- Comments generated: 5
- Review effort level: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
There was a problem hiding this comment.
🔵 Needs a closer look
There are a few correctness gaps (stale Kafka scaler resources not cleaned up on trigger removal, incomplete mutual-TLS wiring for KEDA auth, and an integration test helper still setting legacy scale config) that should be addressed before approval.
Review details
Suppressed comments (6)
Previously missed (5) — in code that hasn't changed since the last review.
pkg/deployer/testing/integration_test_helper.go:746
- This integration test helper still sets scaling under
Deploy.Options.Scale, but the deployers now read scaling from the top-levelFunction.Scale. As written,minScale/maxScalehere will no longer affect the deployed resources, so the test isn’t exercising scale bounds as intended.
pkg/keda/deployer.go:211 - Kafka scaler resources are only created when a Kafka trigger is configured, but there’s no corresponding cleanup when the Kafka trigger is removed (or when switching back to HTTP-only). That can leave a stale
ScaledObject/TriggerAuthenticationbehind and prevent switching triggers cleanly (or keep scaling based on lag unexpectedly).
pkg/keda/kafka_scaling.go:92 needsTriggerAuthdoesn’t consider mutual-TLS inputs (run.kafka.tls.clientCert/clientKey). If a user configures mTLS without a CA cert and without SASL, the ScaledObject won’t reference a TriggerAuthentication at all, so the scaler can’t be given the cert/key material.
This issue also appears on line 178 of the same file.
docs/reference/func_yaml.md:150
- The documented defaults for
scale.min/scale.maxdon’t match the current deployer implementations. For example, the raw deployer defaults to 1 replica, and the keda deployer defaults HTTP scaler bounds to min=1/max=10 when unset.
schema/func_yaml-schema.json:263 schema/func_yaml-schema.jsonis listed as a generated file inAGENTS.mdand shouldn’t be edited directly. Please regenerate it via the repo’s codegen workflow (and commit the generated output) so it stays consistent with the Go struct tags and other generated artifacts.
pkg/keda/kafka_scaling.go:187
buildTriggerAuthonly wiresrun.kafka.tls.caCertinto the TriggerAuthentication. Ifrun.kafka.tls.clientCert/clientKeyare set (mutual TLS), they’re currently ignored, so KEDA can’t be configured with the client cert/key even though the func.yaml schema/docs allow them.
if kafka.TLS != nil && kafka.TLS.CACert != "" {
caSecretName, caKey := findSecretForPath(kafka.TLS.CACert, f.Run.Volumes)
if caSecretName != "" {
secretRefs = append(secretRefs, map[string]interface{}{
"parameter": "ca",
"name": caSecretName,
"key": caKey,
})
}
}
- Files reviewed: 29/29 changed files
- Comments generated: 0 new
- Review effort level: Lite
There was a problem hiding this comment.
🟡 Changes recommended
It introduces a few correctness issues in the KEDA Kafka auth/scaling path and also directly edits a generated schema file that should be regenerated via the project’s codegen workflow.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Review details
Suppressed comments (2)
Previously missed (2) — in code that hasn't changed since the last review.
pkg/functions/function_scale.go:24
- For deployer: keda, allowing scale.max=0 will result in maxReplicaCount=0 in the created HTTPScaledObject/ScaledObject, which is not a valid HPA maxReplicas value. Treat 0 as invalid for keda (or interpret it as "unset" and apply the keda default) to avoid generating invalid autoscaler specs.
schema/func_yaml-schema.json:263 - schema/func_yaml-schema.json is listed as a generated file in AGENTS.md ("Never Do: Edit generated files directly"). Direct edits here are likely to get overwritten and can drift from the generator output; regenerate the schema instead and commit the regenerated result.
- Files reviewed: 29/29 changed files
- Comments generated: 2
- Review effort level: Lite
There was a problem hiding this comment.
🟡 Changes recommended
The keda deploy path can silently proceed with a configured Kafka trigger but missing run.kafka, resulting in a deploy that does not create Kafka scaling resources and does not error.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Review details
Suppressed comments (1)
Previously missed (1) — in code that hasn't changed since the last review.
pkg/keda/kafka_scaling.go:159
- This comment says {{ configMap:... }} refs become a “literal/resolved” KAFKA_SASL_PASSWORD env var, but pkg/k8s/deployer.go actually wires template refs via EnvVar.ValueFrom (secretKeyRef/configMapKeyRef). The wording is misleading for anyone maintaining the TriggerAuthentication wiring.
- Files reviewed: 29/29 changed files
- Comments generated: 1
- Review effort level: Lite
There was a problem hiding this comment.
🔵 Needs a closer look
The KEDA deployer currently produces incorrect/no-HTTP-trigger service URLs and has gaps in deploy-time validation/ScaledObject TLS metadata that can break valid Kafka/KEDA configurations.
Review details
Suppressed comments (3)
Previously missed (3) — in code that hasn't changed since the last review.
pkg/keda/deployer.go:150
- keda.Deployer can be called without going through Function.Validate; in that case, scale.max=0 (or any <1) will be threaded into the HTTPScaledObject/ScaledObject and fail server-side (HPA maxReplicas must be >= 1). Since this deployer already does extra validation for bypass callers (e.g. kafka trigger without run.kafka), it should also fail fast on an invalid scale.max here.
pkg/keda/deployer.go:192 - The in-cluster URL returned for the no-HTTP-trigger case hardcodes port 8080, but the function’s Service listens on port 80 (targetPort 8080). This produces a URL that won’t resolve unless clients also use 8080; other parts of the codebase (e.g. pkg/k8s/describer.go) use the :80 default for Service URLs.
pkg/keda/kafka_scaling.go:302 - Kafka TLS enablement for the KEDA trigger is currently inferred from run.kafka.tls being non-nil. That breaks valid configs that set securityProtocol=SSL/SASL_SSL but rely on system CAs (no explicit run.kafka.tls block): the ScaledObject will omit tls=enable and KEDA will attempt plaintext connections. TLS should be enabled based on run.kafka.securityProtocol instead of presence of the tls struct.
- Files reviewed: 29/29 changed files
- Comments generated: 0 new
- Review effort level: Lite
There was a problem hiding this comment.
🔵 Needs a closer look
There are at least two correctness/documentation issues to address (empty KEDA trigger list can silently deploy without any scaler; run.kafka.tls docs contradict the validated/implemented contract).
Review details
Suppressed comments (2)
Previously missed (2) — in code that hasn't changed since the last review.
pkg/keda/deployer.go:105
- If scale.keda is present but scale.keda.triggers is an explicitly empty list, triggers(f) returns an empty slice, so Deploy skips both the HTTPScaledObject and the Kafka ScaledObject paths and silently falls back to returning a cluster-local Service URL (no KEDA scaling at all). Since Deploy can be reached without Function.Validate, it should fail fast on an empty trigger list instead of deploying a misconfigured function.
docs/reference/func_yaml.md:263 - The docs claim run.kafka.tls is required for securityProtocol SSL/SASL_SSL, but validation only enforces that if tls is set then the protocol must be SSL/SASL_SSL; it allows SSL/SASL_SSL with no tls block (relying on the system trust store), which is also explicitly supported by the KEDA ScaledObject builder. Please adjust the wording so it’s not stricter than the actual config contract.
- Files reviewed: 30/30 changed files
- Comments generated: 0 new
- Review effort level: Lite
There was a problem hiding this comment.
🔵 Needs a closer look
A few edge-case correctness and clarity issues remain (notably around KEDA Kafka SASL mechanism handling and validation messaging) that should be addressed before approval.
Review details
Suppressed comments (3)
Previously missed (3) — in code that hasn't changed since the last review.
pkg/keda/deployer.go:226
- If run.kafka.sasl.mechanism is set to an unexpected value (possible for library callers that bypass Function.Validate), buildScaledObject will set trigger metadata "sasl" to an empty string, which can produce an invalid KEDA trigger configuration. Consider failing fast here with a clear error before creating scaling resources.
pkg/functions/function_scale.go:94 - The default-case error says the allowed trigger types are "http, kafka, cron", but "cron" is explicitly rejected as not yet supported. This is misleading when users mistype the trigger type and then try "cron" based on the message.
pkg/keda/kafka_scaling.go:106 - parseSecretRef uses ad-hoc trimming/splitting and will accept strings that start with "{{" but don’t actually match the project’s template-ref format (and it can diverge from other parsing/validation that uses fn.TemplateRefPattern). Reusing the shared TemplateRefPattern keeps parsing consistent and avoids surprising secret lookups for malformed refs.
- Files reviewed: 30/30 changed files
- Comments generated: 0 new
- Review effort level: Lite
There was a problem hiding this comment.
🟡 Changes recommended
The schema currently contradicts code validation for scale.kpa.target, and the new migration can drop existing scaling config when Function.Root is empty (library callers).
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Review details
Suppressed comments (1)
Previously missed (1) — in code that hasn't changed since the last review.
schema/func_yaml-schema.json:373
- The JSON schema allows scale.kpa.target values down to 0, but runtime validation (ValidateScale/validateKPAScale) requires target >= 0.01. This mismatch will let editors/clients accept invalid func.yaml values that later fail validation; align the schema minimum with the code.
- Files reviewed: 30/30 changed files
- Comments generated: 1
- Review effort level: Lite
There was a problem hiding this comment.
🔵 Needs a closer look
Moderate issues remain in Kafka migration, schema validation, KEDA preflight validation, and scaler cleanup ordering.
Review details
Suppressed comments (4)
Previously missed (1) — in code that hasn't changed since the last review.
pkg/keda/deployer.go:184
- This direct-deployer preflight still omits constraints enforced by
ValidateScale: invalidpollingInterval/cooldownPeriodvalues, per-trigger threshold bounds, and the mutual exclusion ofscale.kedaandscale.kpa. Callers that invoke the deployer withoutFunction.Validatecan therefore create/update the raw Deployment before KEDA rejects or silently ignores an invalid scaler configuration. Reuse the shared validation for explicit scale.keda/kpa configs while preserving the intentional default-HTTP fallback when neither is set.
pkg/functions/function_migrations.go:390
oldDeployonly capturesoptions,deployer, andexpose; it never reads the legacydeploy.kafkafield. A pre-migration func.yaml with Kafka configuration therefore loses it fromRun.Kafkaduring load/migration, so a Kafka trigger will fail validation or deploy without the runtime Kafka settings. Add the legacy field to the migration input and copy it tof.Run.Kafka, with a regression test.
type oldDeploy struct {
Options oldOptions `yaml:"options,omitempty"`
Deployer string `yaml:"deployer,omitempty"`
Expose string `yaml:"expose,omitempty"`
}
pkg/functions/function_options.go:31
ValidateScalerejects an empty trigger list, but the published schema has nominItemsconstraint here, soscale.keda.triggers: []is accepted by schema-based tooling and only fails later during deployment. Add the schema constraint so the declared KEDA contract catches this invalid configuration early.
Triggers []KEDATrigger `yaml:"triggers,omitempty"`
pkg/keda/deployer.go:309
- Scaler reconciliation happens only after
d.Deployer.Deployhas updated the Deployment and waited for availability, and these deletes run only when a trigger type is removed. An existing HTTPScaledObject/ScaledObject with a zero minimum can keep resetting the Deployment to zero while the raw deploy waits for one ready replica, so scale-to-zero functions can time out on redeploys or HTTP↔Kafka transitions before this cleanup runs. Pause/remove the active scaler before the raw update, then recreate or reconcile it after readiness.
if !wantHTTP {
// No HTTP trigger: a prior deploy's HTTPScaledObject and
// interceptor bridge Service, if any, are now orphaned.
if err := deleteHTTPScaledObject(ctx, namespace, f.Name); err != nil {
- Files reviewed: 35/35 changed files
- Comments generated: 0 new
- Review effort level: Lite
There was a problem hiding this comment.
🟡 Changes recommended
Unresolved Kafka-trigger validation, legacy migration and scale compatibility, KEDA readiness, secret-reference handling, and regression coverage remain.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Review details
Suppressed comments (7)
pkg/functions/function_migrations.go:394
- This migration only re-reads
deploy.options.scaleand the legacy observed deployer/exposure fields; it never captures or movesdeploy.kafka. Because the current unmarshal has no destination for that key, an existing function using the legacydeploy.kafkalayout loses its Kafka runtime configuration during migration and will no longer consume Kafka after writing the migrated file. Add the legacy field to the migration input and copy it tof.Run.Kafkawhenrun.kafkais not already set.
var disk struct {
Deploy oldDeploy `yaml:"deploy,omitempty"`
Scale *ScaleOptions `yaml:"scale,omitempty"`
}
pkg/functions/function_scale.go:103
- The Kafka-runtime check is only one-way: it rejects a
kafkatrigger withoutrun.kafka, but it allowsrun.kafkawith a KEDAhttptrigger. That combination passes validation, causes the deployer to create only an HTTPScaledObject, and never creates the Kafka ScaledObject, so the Kafka consumer is not scaled by lag (whilerun.kafkais documented as replacing HTTP serving). Reject this combination or require a Kafka trigger wheneverrun.kafkais used withdeployer: keda.
case "kafka":
sawKafka = true
if kafka == nil {
errors = append(errors, fmt.Sprintf("scale.keda.triggers[%d] has type kafka but run.kafka is not configured", i))
pkg/k8s/deployer.go:677
Options.Scaleremains a public field and is explicitly retained for legacy YAML deserialization, but directFunction/Client.Deploycallers that still populatef.Deploy.Options.Scaleare silently ignored here because deployment generation now reads onlyf.Scale. SinceClient.Deploydoes not run the YAML migration, existing library callers can deploy with default replicas instead of their requested scale. Either use the legacy field as a fallback across deployers or migrate/reject it in the public deployment path.
replicas := int32(1)
if f.Scale != nil && f.Scale.Min != nil && *f.Scale.Min > 0 {
replicas = int32(*f.Scale.Min)
pkg/keda/deployer.go:640
- The new
cooldownPeriodand HTTPtargetValuemapping changes the generatedHTTPScaledObject, but this package has no focused test forhttpScaledObjectwith non-default values; the existing added tests exercise validation and Kafka resources instead. Add a unit test that asserts both configured values are copied into the HTTP scaler so a regression cannot silently restore the hardcoded defaults.
cooldown := int32(300)
targetValue := int64(100)
if scale != nil && scale.KEDA != nil {
if scale.KEDA.CooldownPeriod != nil {
cooldown = *scale.KEDA.CooldownPeriod
}
for _, trig := range scale.KEDA.Triggers {
if trig.Type == "http" && trig.TargetValue != nil {
targetValue = *trig.TargetValue
break
}
pkg/keda/deployer.go:447
ensureScaledObjectreturns as soon as the API create/update succeeds, unlike the HTTP path which waits for the scaler's Ready condition. KEDA can accept this object and then mark it not ready because of invalid trigger metadata, authentication, or operator/RBAC problems; Deploy will still report success even though consumer-lag scaling is not active. Poll the ScaledObject status and surface a non-ready condition before declaring the Kafka deployment successful.
kt := kafkaTrigger(triggers)
so := buildScaledObject(f, kt, deployment, namespace, minScale, maxScale)
if so != nil {
if err := ensureScaledObject(ctx, dynClient, so); err != nil {
return fn.DeploymentResult{}, fmt.Errorf("failed to ensure ScaledObject: %w", err)
pkg/keda/kafka_scaling.go:145
- This returns the first matching volume, so an earlier parent mount such as
/etc/kafkacan shadow a later, more-specific TLS mount at/etc/kafka/ca. The derived key then becomesca/ca.crt, causing the TriggerAuthentication to reference a nonexistent Secret key even though the correct volume is configured. Select the longest matching mount path (or reject overlapping mounts) before returning the Secret reference.
func findSecretForPath(certPath string, volumes []fn.Volume) (secretName, key string) {
for _, v := range volumes {
pkg/keda/lister.go:115
- The new Kafka-only listing path is not covered by the existing KEDA lister integration test: the shared helper deploys the default HTTP-triggered function, while this branch requires an absent
HTTPScaledObjectand a present KafkaScaledObject. Add a focused test for that state, including the fallback readiness and cluster-local URL, so regressions in the dynamic-resource lookup are detected.
if !hasHTTPTrigger {
// Absence of an HTTPScaledObject isn't proof this is a Kafka-only
// function -- an http-triggered function's scaler could have been
// deleted externally, failed to create, or be mid-transition.
// Corroborate with the Kafka ScaledObject before assuming
// Kafka-only; if neither scaler exists, something's broken and
// should be reported instead of silently listed as healthy.
if _, err := dynClient.Resource(scaledObjectGVR).Namespace(namespace).Get(ctx, scaledObjectName(name), metav1.GetOptions{}); err != nil {
if errors.IsNotFound(err) {
return fn.ListItem{}, fmt.Errorf(
"function %q uses the keda deployer but has neither an HTTPScaledObject nor a Kafka ScaledObject: the scaler may have failed to create or been deleted externally", name)
}
return fn.ListItem{}, fmt.Errorf("unable to get ScaledObject: %v", err)
}
}
- Files reviewed: 35/35 changed files
- Comments generated: 1
- Review effort level: Lite
32e3c8f to
5fc56ca
Compare
There was a problem hiding this comment.
🔵 Needs a closer look
Unresolved migration and raw-deployer scale validation issues remain, and the documentation lists an unsupported cron trigger.
Review details
Suppressed comments (4)
docs/reference/func_yaml.md:165
- The documentation presents
cronas a usable KEDA trigger, butvalidateKEDAScaleexplicitly rejects it as not yet supported and the deployer only materializes HTTP and Kafka triggers. A user following this section gets a validation error; either implement cron or clearly mark/remove these options until it is supported.
- `triggers`: a list of KEDA triggers. At least one is required. Each trigger has a `type` of `http`, `kafka`, or `cron`:
- `http`: scales based on incoming HTTP request rate.
- `targetValue`: requests per second per replica before scaling up. Default is 100.
- `kafka`: scales based on consumer group lag. Requires [`run.kafka`](#runkafka) to be configured.
- `lagThreshold`: average consumer lag per partition that triggers scaling up. Default is 10.
pkg/functions/function_migrations.go:500
- At migration time
f.Deployeris still the value read from func.yaml; command/global overrides are applied later (cmd/deploy.go:313). For an old file with flatdeploy.options.scaleand no persisted deployer, selecting--deployer raworkedaleavesvalidKPADeployertrue, createsscale.kpa, and skips the KEDA HTTP default; validation then rejects the requested deployer. Defer this conversion/defaulting until the effective deployer is known, or preserve the legacy settings without binding them to KPA.
validKPADeployer := f.Deployer == "" || f.Deployer == "knative"
if hasFlat && newScale.KPA == nil && validKPADeployer {
pkg/functions/function_migrations.go:444
- This migration never decodes the legacy
deploy.kafkafield or copies it intof.Run.Kafka. Because the normal YAML unmarshal ignores that old key, upgrading an existing Kafka function drops its consumer configuration: Kafka-triggered functions then fail validation, while fixed/HTTP-scaled consumers are deployed without Kafka. Decode the legacy field and move it torun.kafka(while preserving an already-present top-level value), and add a migration regression test.
var disk struct {
Deploy oldDeploy `yaml:"deploy,omitempty"`
Scale *ScaleOptions `yaml:"scale,omitempty"`
}
pkg/k8s/deployer.go:677
- This path now casts the top-level
scale.mindirectly toint32, butClient.Deploydoes not callFunction.Validate, so direct raw-deployer callers can pass values outside the Kubernetes range and get wraparound (for example2147483648becomes zero) instead of an error. Add the same bounds check used by the KEDA direct-deploy preflight before narrowing this value.
replicas := int32(1)
if f.Scale != nil && f.Scale.Min != nil && *f.Scale.Min > 0 {
replicas = int32(*f.Scale.Min)
- Files reviewed: 35/35 changed files
- Comments generated: 0 new
- Review effort level: Lite
There was a problem hiding this comment.
🔵 Needs a closer look
Six moderate findings remain unresolved across migration, schema, validation, lifecycle ordering, and pipeline deployer handling.
Review details
Suppressed comments (6)
pkg/functions/function_migrations.go:417
- This 0.38.0 migrator never reads a legacy
deploy.kafkafield; it only captures the old scale and deployer/expose fields. Since the currentFunctionno longer has aDeploySpec.Kafkafield, YAML unmarshalling and this disk-side struct both discard that data, so existing functions using the promised old Kafka location lose their runtime configuration instead of being moved torun.kafka. Add the legacy field to the migration input and copy it intof.Run.Kafka, with a migration test.
// migrateScaleToTopLevel moves scale config from deploy.options.scale to the
// top-level scale field, moves the flat metric/target/utilization fields
// (from pre-0.38.0 func.yaml files) into the kpa sub-key, and renames
// DeploySpec's observed-state deploy.deployer/deploy.expose YAML keys to
// deploy.activeDeployer/deploy.activeExpose (they collided in name with the
// top-level deployer/expose fields -- user intent -- despite meaning the
// opposite thing: observed, currently-deployed state).
func migrateScaleToTopLevel(f Function, m migration) (Function, error) {
pkg/functions/function_scale.go:33
- The new int32 upper-bound validation is not represented in the generated schema:
scale.minandscale.maxstill accept any non-negative integer, so values such as2147483648pass schema validation but are rejected here (and would not fit Kubernetes replica counts). Add the same maximum to the schema source/generator and regenerateschema/func_yaml-schema.json.
if scale.Min != nil && *scale.Min > math.MaxInt32 {
errors = append(errors, fmt.Sprintf("scale.min has invalid value: %d, must be <= %d", *scale.Min, math.MaxInt32))
}
if scale.Max != nil && *scale.Max > math.MaxInt32 {
errors = append(errors, fmt.Sprintf("scale.max has invalid value: %d, must be <= %d", *scale.Max, math.MaxInt32))
pkg/k8s/deployer.go:677
- Because the raw deployer is callable without
Function.Validate, a negativescale.minreaches this condition, skips the block, and is silently normalized to one replica. That violates the non-negative scale contract and is inconsistent with the overflow guard added here; reject values< 0as well, while keeping the initial replica fallback for exactly zero.
if f.Scale != nil && f.Scale.Min != nil && *f.Scale.Min > 0 {
pkg/keda/deployer.go:439
- When a valid Kafka configuration is changed from SASL/TLS credentials to no authentication, this deletes the existing
TriggerAuthenticationbeforeensureScaledObjectupdates the existingScaledObjectto remove itsauthenticationRef. If that update is rejected or fails transiently, the old scaler continues referencing a deleted authentication resource and Kafka scaling is broken. Reconcile theScaledObjectfirst, then delete the now-unreferencedTriggerAuthenticationonly after that update succeeds.
if err := deleteTriggerAuth(ctx, dynClient, namespace, triggerAuthName(f.Name)); err != nil {
fmt.Fprintf(os.Stderr, "warning: %v\n", err)
pkg/pipelines/tekton/pipelines_provider.go:161
- The effective deployer is only copied into
Deploy.ActiveDeployer, leavingf.Deployerempty when a flag-less pipeline call relies on the persisted active deployer. For a previously deployed KEDA function this makes the laterf.Validate()rejectscale.kedawithrequires deployer: keda, even thoughfunc-utilwill select KEDA fromActiveDeployer; for an active raw function, KPA settings can instead pass validation and then be silently ignored. Copy the resolved deployer into the intent field (or validate against the resolved value) before validation and serializing the pipeline input.
schema/func_yaml-schema.json:285 scale.keda.triggersis documented and runtime-validated as required, but this generated schema only setsminItemswhen the property is present and has norequiredentry. Consequentlyscale: {keda: {}}passes schema validation whileValidateScaleand the KEDA deployer reject it. Make the source/generator emitrequired: ["triggers"]and regenerate the schema.
- Files reviewed: 35/35 changed files
- Comments generated: 0 new
- Review effort level: Lite
There was a problem hiding this comment.
🔵 Needs a closer look
Unresolved migration, scaler-transition cleanup, exposure, and schema-validation issues block approval.
Review details
Suppressed comments (6)
pkg/functions/function_migrations.go:444
- The PR description says existing configurations with Kafka under
deploy.kafkaare migrated torun.kafka, but this migration only readsdeploy.options.scale,deploy.deployer, anddeploy.expose. Since the current Function shape has nodeploy.kafkafield, such a legacy config is silently dropped during unmarshal and the resulting function is no longer a Kafka consumer. Add an explicit legacy Kafka field to the migration input and copy it tof.Run.Kafkabefore writing the migrated function.
type oldDeploy struct {
Options oldOptions `yaml:"options,omitempty"`
Deployer string `yaml:"deployer,omitempty"`
Expose string `yaml:"expose,omitempty"`
}
var disk struct {
Deploy oldDeploy `yaml:"deploy,omitempty"`
Scale *ScaleOptions `yaml:"scale,omitempty"`
}
pkg/functions/function_migrations.go:500
- Migration runs before cmd/deploy applies the CLI or global
--deployeroverride. For a legacy file with flatdeploy.options.scaleand no persisted top-level deployer, this branch sees an emptyf.Deployerand moves the fields intoscale.kpa; a later--deployer=kedathen makesValidateScalereject the migrated function. Please defer this deployer-dependent classification until after the effective deployer is resolved (and add a CLI/global-override regression test).
validKPADeployer := f.Deployer == "" || f.Deployer == "knative"
if hasFlat && newScale.KPA == nil && validKPADeployer {
pkg/keda/deployer.go:341
- A failed deletion is only logged before the HTTP scaler path proceeds. Because the Deployment remains in place when switching from Kafka to HTTP, ownerReferences do not clean up this ScaledObject; the subsequent HTTPScaledObject can therefore conflict with the stale Kafka scaler or leave both controllers scaling the workload. Treat non-NotFound cleanup failures as fatal, or wait until the old scaler is actually gone before creating the HTTP scaler.
if err := deleteScaledObject(ctx, dynClient, namespace, scaledObjectName(f.Name)); err != nil {
fmt.Fprintf(os.Stderr, "warning: %v\n", err)
} else if d.verbose {
pkg/keda/deployer.go:327
- If an existing HTTPScaledObject cannot be deleted (for example because the deployer lacks permission), this warning is ignored and the Kafka scaler is created anyway. The Deployment is retained during a trigger transition, so its owner reference cannot garbage-collect that stale HTTP scaler; KEDA can then reject the new scaler or have two scalers competing for the same Deployment. Treat non-NotFound cleanup failures as a deploy error, or wait until the old scaler is actually gone before provisioning the replacement.
if err := deleteHTTPScaledObject(ctx, namespace, f.Name); err != nil {
fmt.Fprintf(os.Stderr, "warning: %v\n", err)
} else if d.verbose {
pkg/keda/deployer.go:406
- When a Kafka-only function requests
expose: route, this branch skips the configured KEDA exposer and never records an applied exposure.Function.Exposeexplicitly applies to the KEDA deployer, so an OpenShift deploy can report success while silently ignoring the user's route request; reject this combination before creating resources or implement a Kafka-only exposure path.
} else {
// No HTTP trigger — URL is the app service. The Service listens on
// port 80 (routing to the container's DefaultHTTPPort via
// targetPort), so the URL, like elsewhere in the codebase (e.g.
// pkg/k8s/describer.go), has no explicit port.
url = fmt.Sprintf("http://%s.%s.svc", f.Name, namespace)
// A prior deploy may have exposed this function over HTTP. Nothing
// reconciles that exposure once the HTTP trigger is gone, so clear
// it the same way deployClusterLocal does -- otherwise the old
// Route and the Service's exposure annotations stay active,
// pointing at a function that no longer has anything serving HTTP.
target := deployTarget{
clientset: k8sClientset,
dynClient: dynClient,
ref: deployer.NewExposureRef(f.Name, namespace, ""),
}
if err := d.clearExposure(ctx, target, appService.Annotations[k8s.RouteNamespaceAnnotation]); err != nil {
return fn.DeploymentResult{}, err
}
schema/func_yaml-schema.json:364
- The generated schema now represents
scale.kpa.targetasexclusiveMinimum: truewithminimum: 0, so values such as0.005pass schema validation even thoughValidateScalerejects every value below0.01. This makes schema validation and runtime validation disagree; update the generator metadata/generation path so the documented decimal lower bound is enforced consistently.
- Files reviewed: 36/36 changed files
- Comments generated: 0 new
- Review effort level: Lite
|
/retest-required |
gauron99
left a comment
gauron99
left a comment
There was a problem hiding this comment.
First pass on the PR. Thanks for doing the migration and the scale yaml movement too!
Im not sure if my page is messed up but im seeing a lot of ActiveDeployer left over after that 4th commit -- Ive stopped naming them in the reviews after a couple.
Ive also though we could simplify the PR by removing the deployer migration (which seems unrelated and its there as a transient "fix") -> we fix the empty scale validator -> we can remove the deployer migration
| // decision and the keda-defaults-to-http-trigger block see it. Read from | ||
| // disk rather than f.Deploy.ActiveDeployer so the migration is robust to | ||
| // callers that pass a Function not populated via the primary unmarshal. | ||
| if f.Deployer == "" && disk.Deploy.Deployer != "" { |
There was a problem hiding this comment.
I think if we fix the validation (to allow empty scale because default should be the http scaler as is today on main) we can completely remove this deploy.deployer -> deployer intent transition.
This is technically fixing one scenario if im not missing any and that is if a user has a function deployed and they run func delete with a new binary the deploy.deployer (which was the only key for them but now is status only) would get removed and the intent would be lost so re-running a func deploy would run with defaultDeployer instead of their declared one. I think that blast radius is extremely small and it does not need to be coupled with this PR
There was a problem hiding this comment.
Agreed — done. Removed the deploy.deployer → deployer intent-recovery block entirely. The remaining edge you noted (a func delete with a newer binary dropping the intent so a re-deploy falls back to the default deployer) is pre-existing on main and decoupled from this PR as you suggested — tracked in #4054.
| // deployer and Kafka config. It replaces the previous validateScaleDeployer, | ||
| // validateKEDAScale, and validateKPAScale functions with a single entry point. | ||
| func ValidateScale(scale *ScaleOptions, deployer string, kafka *KafkaConfig) (errors []string) { | ||
| if deployer == "keda" && (scale == nil || scale.KEDA == nil || len(scale.KEDA.Triggers) == 0) { |
There was a problem hiding this comment.
I think if we go with the rest of fixes we can drop this condition block and say that empty scale is valid for every deployer -> for keda deployer it simply means "use http scaler"
There was a problem hiding this comment.
Done — dropped the block. scale == nil now returns valid for every deployer; for deployer: keda it means "use the http scaler", supplied by the deployer's triggers() helper. Only an explicitly-written scale.keda with an empty triggers: [] is still rejected.
| // callers, tests): without this check, Deploy would silently skip | ||
| // both the HTTPScaledObject and Kafka ScaledObject paths and deploy | ||
| // with no scaler at all. | ||
| return fn.DeploymentResult{}, fmt.Errorf("function %q: deployer keda requires at least one trigger in scale.keda.triggers", f.Name) |
There was a problem hiding this comment.
as mentioned above in some test i reviewed first I think we should build-in a default to always fallback to http scaler (as is currently on main now) when the scale.keda is nil (not explicitly [] empty) & deployer is keda.
This will give us backwards compatibility and super ease of use for users because you dont need to worry about anything after func deploy --deployer keda.
There was a problem hiding this comment.
Done — deployer: keda with a nil scale.keda now defaults to a single http trigger (backwards-compatible with main). func deploy --deployer keda works with no scale config at all; only an explicitly-written empty triggers: [] is rejected. Docs (func_yaml.md) updated to state this.
| @@ -99,8 +102,165 @@ func (k *kedaDeployerDecorator) UpdateLabels(function fn.Function, labels map[st | |||
| } | |||
|
|
|||
| func (d *Deployer) Deploy(ctx context.Context, f fn.Function) (fn.DeploymentResult, error) { | |||
There was a problem hiding this comment.
the deployer seems to be doing a lot of validation that is already done beforehand duplicating. Also the deployer should be minimal "executor". I think we could trim this down to nil guards and name validators (for k8s resource char limits) and secret path?
There was a problem hiding this comment.
Agreed the deployer is doing too much and should be a minimal executor — but Id like to keep it for this PR and do the trim separately, tracked in #4055.
The reason it can't just be deleted now: these checks aren't re-validation within one process — each guards a different deploy process, and Function.Validate() (which runs ValidateScale) is skipped in most of the flows that reach the deployer. func-util deploy (the on-cluster Tekton deploy step) calls client.Deploy(ctx, f) with no Validate(), and Client.Deploy itself only validates expose/switch, never scale. So for every on-cluster deploy (and any library caller) the deployer's preflight is currently the only validation that runs.
The genuinely deployer-only bits you called out — k8s resource-name length and secret/TLS paths — stay regardless. The rest duplicates ValidateScale/ValidateKafkaSecurity, and the clean fix is to validate once at the shared choke point (Client.Deploy, which both the CLI and func-util go through) and then shrink the deployer. That's a cross-deployer change with a scoping decision (deploy-relevant validation vs. the full authoring Validate()), so it's better as its own PR — #4055.
ccecad9 to
0ff075b
Compare
There was a problem hiding this comment.
🟡 Changes recommended
Unresolved moderate migration/schema issues and critical KEDA cleanup failures block approval.
Get a fresh assessment by requesting another Copilot review.
Review details
Suppressed comments (2)
docs/reference/func_yaml.md:156
- This new scale documentation points to
options.resources.limits.concurrency, butresourcesis serialized underdeploy.options(as shown by the example immediately below). A user following this path would write a top-leveloptionsblock that the schema rejects; document the actualdeploy.options.resources.limits.concurrencypath.
- `target`: target value for the metric. Defaults to `options.resources.limits.concurrency` when given. Float >= 0.01, default is 100. See related [Knative docs](https://knative.dev/docs/serving/autoscaling/concurrency/#soft-limit).
pkg/functions/function_migrations.go:519
- The migration decides whether to create
scale.kpaand whether to inject the default HTTP trigger fromf.Deployer, but older files can carry only the observed deployer indeploy.deployer. The code copies that value intof.Deploy.Deployerlater (lines 536-540), so a legacy KEDA file reaches validation with either a misplacedscale.kpaor an emptyscale.keda.triggersand fails/loses the promised HTTP default. Use the effective deployer from the legacy observed field as a migration-only fallback without changing the intent field.
if f.Deployer == "keda" {
if f.Scale == nil {
f.Scale = &ScaleOptions{}
}
if f.Scale.KEDA == nil {
- Files reviewed: 31/32 changed files
- Comments generated: 5
- Review effort level: Lite
…r transition When a function switches trigger type (http<->kafka), the keda deployer deletes the other type's stale scaler before creating the new one. Those deletions were best-effort (logged as warnings), so a non-NotFound delete failure left two scalers targeting the same Deployment while Deploy still reported success. Unlike Remover.Remove, a transition keeps the Deployment, so owner-reference GC does not clean up the leftover scaler. Return the error instead. The delete helpers already suppress NotFound, so steady-state single-type deploys stay a no-op; only a real API failure mid -transition now aborts. Add unit coverage that the delete helpers propagate non-NotFound errors.
|
/retest-required |
|
this seems to be an issue |
…ources The trigger-transition cleanup made all four stale-resource deletions fatal, but two of them (interceptor bridge Service, TriggerAuthentication) are not scalers and run speculatively on every deploy. On-cluster the deploy ServiceAccount may lack the delete verb for a resource that never existed; Kubernetes authorizes before checking existence, so that delete returns Forbidden (not NotFound) and aborted every on-cluster http keda deploy (TestInt_Remote_Default/keda). Narrow fatality to the actual scalers (HTTPScaledObject, ScaledObject), which are the only resources that trip KEDA's one-scaler-per-workload rule. Their companions are inert when orphaned and now log a warning instead of failing the deploy.
Changes
This PR adds Kafka consumer-lag autoscaling support via KEDA. When
deployer: kedais set and a Kafka trigger is configured, the deployer creates a KEDA ScaledObject with a Kafka trigger that scales function pods based on consumer group lag, along with a TriggerAuthentication resource for SASL/TLS credentials.The func.yaml scale configuration has been restructured:
scaleis now a top-level field (previously nested underdeploy.options.scale), withmin/maxplus one of two mutually-exclusive sub-keys:scale.kpafor theknativedeployer:metric,target,utilization.scale.kedafor thekedadeployer:pollingInterval,cooldownPeriod, and atriggerslist. Each trigger (http,kafka, orcron) carries its own knobs, e.g.lagThresholdandactivationLagThresholdfor Kafka triggers.run.kafka(brokers, topic, consumerGroup, security protocol, SASL/TLS settings).A migration handles existing func.yaml files automatically: it moves
deploy.options.scaleto the top-levelscale, lifts the old flatmetric/target/utilizationfields intoscale.kpa, and renamesdeploy.deployertodeploy.activeDeployer.The PR also includes testing deployment scenarios (A-G) with collected Kubernetes resource snapshots for each deployer/Kafka combination.
/kind enhancement
Supersedes #4045
Closes #2419
Release Note
Docs