Merge branch 'main' into feat/flow-control-default-on

Author: ahg-g

.github/workflows/ci-lint.yaml

@@ -46,6 +46,7 @@ jobs:
               - go.mod
               - go.sum
               - .golangci.yml
+              - .golangci-security.yml
               - .typos.toml
               - scripts/**
               - hack/**
@@ -57,6 +58,10 @@ jobs:
     if: ${{ needs.check-changes.outputs.src == 'true' }}
     runs-on: ubuntu-latest
     timeout-minutes: 30
+    permissions:
+      contents: read
+      actions: read
+      security-events: write
     steps:
       - name: Checkout source
         uses: actions/checkout@v7
@@ -103,3 +108,28 @@ jobs:
         run: |
           go install golang.org/x/vuln/cmd/govulncheck@v1.3.0
           govulncheck ./...
+
+      - name: Run make lint-security
+        env:
+          GO_MOD_CACHE_VOL: ${{ steps.go-cache.outputs.mod }}
+          GO_BUILD_CACHE_VOL: ${{ steps.go-cache.outputs.build }}
+        run: make lint-security
+
+      # Fork PRs run with a read only GITHUB_TOKEN that cannot upload SARIF.
+      - name: Upload gosec SARIF
+        if: ${{ !cancelled() && (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) }}
+        uses: github/codeql-action/upload-sarif@68bde559dea0fdcac2102bfdf6230c5f70eb485e # v4.35.4
+        with:
+          sarif_file: gosec.sarif
+          category: gosec
+
+      # On runs that skip the upload above the artifact is the only way to
+      # reach the findings.
+      - name: Upload gosec SARIF artifact
+        if: ${{ !cancelled() }}
+        uses: actions/upload-artifact@v7
+        with:
+          name: gosec-sarif
+          path: gosec.sarif
+          retention-days: 7
+          if-no-files-found: warn

.gitignore

@@ -19,6 +19,9 @@ bin/
 # Output of the go coverage tool, specifically when used with LiteIDE
 *.out
 
+# Security scan report written by `make lint-security`
+gosec.sarif
+
 # Dependency directories (remove the comment below to include it)
 # vendor/
 

.golangci-security.yml

@@ -0,0 +1,21 @@
+version: "2"
+
+# Security linters only. Kept separate from .golangci.yml so the SARIF report
+# contains gosec findings alone and uploads to the code scanning "gosec"
+# category without style findings mixed in.
+
+run:
+  timeout: 5m
+  allow-parallel-runners: true
+
+linters:
+  default: none
+  enable:
+    - gosec
+
+output:
+  formats:
+    text:
+      path: stdout
+    sarif:
+      path: gosec.sarif

Makefile

@@ -260,6 +260,15 @@ lint: image-build-builder ## Run lint (use LINT_NEW_ONLY=true to only check new
 	@printf "\033[33;1m==== Running linting ====\033[0m\n"
 	$(BUILDER_RUN) 'GOFLAGS=-buildvcs=false golangci-lint run $(LINT_ARGS) && typos'
 
+# Reports findings without failing while the initial baseline is triaged.
+# Set to 1 to block merges on new findings.
+SECURITY_LINT_EXIT_CODE ?= 0
+
+.PHONY: lint-security
+lint-security: image-build-builder ## Run security linters and write gosec.sarif
+	@printf "\033[33;1m==== Running security linting ====\033[0m\n"
+	$(BUILDER_RUN) 'GOFLAGS=-buildvcs=false golangci-lint run --config=./.golangci-security.yml --issues-exit-code=$(SECURITY_LINT_EXIT_CODE)'
+
 .PHONY: test
 test: test-unit test-e2e ## Run all tests (unit and e2e)
 

apix/config/v1alpha1/endpointpickerconfig_types.go

@@ -240,13 +240,25 @@ type DataLayerConfig struct {
 	// endpoints. This enables running the EPP without a Kubernetes cluster.
 	// If omitted, the EPP uses the default Kubernetes-based discovery.
 	Discovery *DiscoveryConfig `json:"discovery,omitempty"`
+	// +optional
+	// CrossReplicaSyncerPluginRef names the plugin instance to use as the cross-EPP
+	// cross-replica syncer. The reference is to the name of an entry in the
+	// top-level Plugins section. If omitted, no cross-replica syncer is used
+	// and plugins that read cross-replica state fall back to local data.
+	CrossReplicaSyncerPluginRef string `json:"crossReplicaSyncerPluginRef,omitempty"`
+	// +optional
+	// CrossReplicaSyncInterval is the cadence at which each replica publishes
+	// its local per-endpoint state to the cross-replica syncer. It is rounded
+	// to a multiple of the datalayer base tick. If omitted, a default is used.
+	CrossReplicaSyncInterval *metav1.Duration `json:"crossReplicaSyncInterval,omitempty"`
 }
 
 func (dlc *DataLayerConfig) String() string {
 	if dlc == nil {
 		return nilString
 	}
-	return fmt.Sprintf("{Sources: %v, Discovery: %v}", dlc.Sources, dlc.Discovery)
+	return fmt.Sprintf("{Sources: %v, Discovery: %v, CrossReplicaSyncerPluginRef: %s, CrossReplicaSyncInterval: %v}",
+		dlc.Sources, dlc.Discovery, dlc.CrossReplicaSyncerPluginRef, dlc.CrossReplicaSyncInterval)
 }
 
 // DiscoveryConfig references the EndpointDiscovery plugin to use.

apix/config/v1alpha1/zz_generated.deepcopy.go

@@ -472,6 +472,7 @@ func (r *Runner) setup(ctx context.Context, cfg *rest.Config, opts *runserver.Op
 		GRPCMaxRecvMsgSize:               opts.GRPCMaxRecvMsgSize,
 		GRPCMaxSendMsgSize:               opts.GRPCMaxSendMsgSize,
 		EnableGRPCStreamMetrics:          opts.EnableGRPCStreamMetrics,
+		EmitEndpointScores:               opts.EmitEndpointScores,
 	}
 
 	if err := serverRunner.SetupWithManager(mgr); err != nil {
@@ -706,7 +707,7 @@ func makePodListFunc(ds datastore.Datastore) func() []types.NamespacedName {
 		names := make([]types.NamespacedName, 0, len(pods))
 
 		for _, p := range pods {
-			names = append(names, p.GetMetadata().NamespacedName)
+			names = append(names, p.GetMetadata().ID)
 		}
 		return names
 	}
@@ -995,6 +996,7 @@ func (r *Runner) runWithFileDiscovery(ctx context.Context, opts *runserver.Optio
 		GRPCMaxRecvMsgSize:               opts.GRPCMaxRecvMsgSize,
 		GRPCMaxSendMsgSize:               opts.GRPCMaxSendMsgSize,
 		EnableGRPCStreamMetrics:          opts.EnableGRPCStreamMetrics,
+		EmitEndpointScores:               opts.EmitEndpointScores,
 	}
 
 	r.customCollectors = append(r.customCollectors, collectors.NewInferencePoolMetricsCollector(ds))

cmd/epp/runner/runner.go

@@ -105,7 +105,7 @@ type EndpointDiscovery interface {
 type DiscoveryNotifier interface {
     // Upsert adds or updates an endpoint in the datastore.
     Upsert(endpoint *EndpointMetadata)
-    // Delete removes an endpoint by its namespaced name.
+    // Delete removes an endpoint by its ID.
     Delete(id types.NamespacedName)
 }
 ```
@@ -114,8 +114,8 @@ type DiscoveryNotifier interface {
 
 | Field | Type | Description |
 |---|---|---|
-| `NamespacedName` | `types.NamespacedName` | Unique identity of the endpoint. |
-| `PodName` | `string` | Logical name (used in metrics). |
+| `ID` | `types.NamespacedName` | Unique identity of the endpoint. Each discovery source must set it uniquely across the endpoints it reports. |
+| `Name` | `string` | Name of the workload behind the endpoint. |
 | `Address` | `string` | IP address of the inference server. |
 | `Port` | `string` | Port as a string (e.g. `"8000"`). |
 | `MetricsHost` | `string` | `host:port` for metrics scraping. Defaults to `address:port` if empty. |
@@ -555,9 +555,9 @@ func (d *MyDiscovery) TypedName() fwkplugin.TypedName { return d.typedName }
 func (d *MyDiscovery) Start(ctx context.Context, notifier fwkdl.DiscoveryNotifier) error {
     // 1. Enumerate existing endpoints.
     notifier.Upsert(&fwkdl.EndpointMetadata{
-        NamespacedName: types.NamespacedName{Name: "ep0", Namespace: "default"},
-        Address:        "10.0.0.1",
-        Port:           "8000",
+        ID:      types.NamespacedName{Name: "ep0", Namespace: "default"},
+        Address: "10.0.0.1",
+        Port:    "8000",
         MetricsHost:    "10.0.0.1:8000",
     })
 

docs/discovery.md

@@ -447,6 +447,17 @@ func buildDataLayerConfig(rawDataConfig *configapi.DataLayerConfig, handle fwkpl
 		return &cfg, nil
 	}
 
+	if ref := rawDataConfig.CrossReplicaSyncerPluginRef; ref != "" {
+		syncer, ok := handle.Plugin(ref).(fwkdl.CrossReplicaSyncer)
+		if !ok {
+			return nil, fmt.Errorf("the plugin %s is not a fwkdl.CrossReplicaSyncer", ref)
+		}
+		cfg.Syncer = syncer
+	}
+	if iv := rawDataConfig.CrossReplicaSyncInterval; iv != nil {
+		cfg.SyncInterval = iv.Duration
+	}
+
 	for _, source := range rawDataConfig.Sources {
 		if sourcePlugin, ok := handle.Plugin(source.PluginRef).(fwkdl.DataSource); ok {
 			sourceConfig := datalayer.DataSourceConfig{

pkg/epp/config/loader/configloader.go

@@ -204,7 +204,7 @@ func diffStore(store datastore.Datastore, params diffStoreParams) string {
 	}
 	gotEndpoints := []string{}
 	for _, em := range store.PodList(datastore.AllPodsPredicate) {
-		gotEndpoints = append(gotEndpoints, em.GetMetadata().NamespacedName.Name)
+		gotEndpoints = append(gotEndpoints, em.GetMetadata().ID.Name)
 	}
 	if diff := cmp.Diff(params.wantEndpoints, gotEndpoints, cmpopts.SortSlices(func(a, b string) bool { return a < b })); diff != "" {
 		return "endpoints:" + diff