From 00ea163b4f3ea9443e64d66ff2f35333341a06c3 Mon Sep 17 00:00:00 2001 From: Ikhun Um Date: Fri, 7 Aug 2026 10:38:29 +0900 Subject: [PATCH 1/5] Reject stale K8s aggregator updates and let the policy through Motivation: Two aggregator edits that overlap in time both succeeded: the second one was built from a document read before the first one landed, so it rolled that change back with no sign anything had happened. Separately, the load-balancing policy on an aggregator was stored but never reached the endpoints generated from it, so setting it did nothing. Modifications: - The update endpoint takes the revision the client read the aggregator at and uses it as the base revision of the commit, so Central Dogma rejects the push itself: a conflict is 409, an unknown revision 400. Central Dogma compares revisions per repository, so any commit in the group makes an open editor stale; omitting the parameter keeps the previous always-apply behaviour. Mapping ChangeConflictException to 409 is shared by every xDS resource type, which previously answered 500. - The policy is copied into the generated ClusterLoadAssignment and into the preview, and is now validated on create and update: Envoy applies at most one drop overload and rejects the whole assignment when it sees more, or an overprovisioning factor of zero. Aggregators that already store such a policy were harmless while it was ignored and would now be served, so they are worth checking before this ships. Result: An aggregator edit based on a stale read is refused instead of silently overwriting a newer one, and a policy set on an aggregator takes effect. --- .../xds/internal/XdsResourceManager.java | 39 ++++- .../XdsKubernetesEndpointFetchingService.java | 3 + .../xds/k8s/v1/XdsKubernetesService.java | 50 +++++- .../xds/k8s/v1/XdsKubernetesServiceTest.java | 162 ++++++++++++++++++ 4 files changed, 245 insertions(+), 9 deletions(-) diff --git a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java index 9a745493b..e30ee665c 100644 --- a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java +++ b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java @@ -45,9 +45,11 @@ import com.linecorp.armeria.common.util.Exceptions; import com.linecorp.centraldogma.common.Author; import com.linecorp.centraldogma.common.Change; +import com.linecorp.centraldogma.common.ChangeConflictException; import com.linecorp.centraldogma.common.Markup; import com.linecorp.centraldogma.common.RedundantChangeException; import com.linecorp.centraldogma.common.Revision; +import com.linecorp.centraldogma.common.RevisionNotFoundException; import com.linecorp.centraldogma.internal.Jackson; import com.linecorp.centraldogma.internal.Yaml; import com.linecorp.centraldogma.server.command.Command; @@ -232,13 +234,21 @@ public CompletableFuture push( private CompletableFuture doPush( String group, String fileName, String summary, Author author, boolean create, @Nullable String legacyFileToRemove, String originalBody) { + return doPush(group, fileName, summary, author, create, legacyFileToRemove, originalBody, + Revision.HEAD); + } + + private CompletableFuture doPush( + String group, String fileName, String summary, + Author author, boolean create, @Nullable String legacyFileToRemove, String originalBody, + Revision baseRevision) { // Store the original YAML body as-is (server-set fields are injected by the caller before // this method is invoked). Respond with the same body so the client sees exactly what is stored. final Change change = Change.ofYamlUpsert(fileName, originalBody); final ImmutableList> changes = legacyFileToRemove != null ? ImmutableList.of(Change.ofRemoval(legacyFileToRemove), change) : ImmutableList.of(change); - return commandExecutor.execute(Command.push(author, INTERNAL_PROJECT_XDS, group, Revision.HEAD, + return commandExecutor.execute(Command.push(author, INTERNAL_PROJECT_XDS, group, baseRevision, summary, "", Markup.PLAINTEXT, changes)) .handle((unused, cause) -> { if (cause != null) { @@ -246,6 +256,13 @@ private CompletableFuture doPush( if (!create && peeled instanceof RedundantChangeException) { return toYamlResponse(originalBody); } + if (peeled instanceof ChangeConflictException) { + return errorResponse(HttpStatus.CONFLICT, peeled); + } + if (peeled instanceof RevisionNotFoundException) { + return errorResponse(HttpStatus.BAD_REQUEST, + "Invalid revision: " + baseRevision); + } return errorResponse(HttpStatus.INTERNAL_SERVER_ERROR, peeled); } return toYamlResponse(originalBody); @@ -254,16 +271,32 @@ private CompletableFuture doPush( public CompletableFuture update( String group, String resourceName, String summary, Author author, String originalBody) { - return update(group, resourceName, fileName(group, resourceName), summary, author, originalBody); + return update(group, resourceName, summary, author, originalBody, Revision.HEAD); + } + + // A base revision other than HEAD makes the commit a compare-and-swap: it is rejected with 409 once the + // group repository has advanced past it, so a save based on a stale read cannot overwrite a newer one. + public CompletableFuture update( + String group, String resourceName, String summary, Author author, String originalBody, + Revision baseRevision) { + return update(group, resourceName, fileName(group, resourceName), summary, author, originalBody, + baseRevision); } public CompletableFuture update( String group, String resourceName, String fileName, String summary, Author author, String originalBody) { + return update(group, resourceName, fileName, summary, author, originalBody, Revision.HEAD); + } + + private CompletableFuture update( + String group, String resourceName, String fileName, String summary, + Author author, String originalBody, Revision baseRevision) { return updateOrDelete(group, resourceName, fileName, resolvedFileName -> { final String legacyFileToRemove = resolvedFileName.endsWith(".json") ? resolvedFileName : null; final String targetFileName = legacyFileToRemove != null ? fileName : resolvedFileName; - return doPush(group, targetFileName, summary, author, false, legacyFileToRemove, originalBody); + return doPush(group, targetFileName, summary, author, false, legacyFileToRemove, originalBody, + baseRevision); }); } diff --git a/xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesEndpointFetchingService.java b/xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesEndpointFetchingService.java index dc5556423..9be05274a 100644 --- a/xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesEndpointFetchingService.java +++ b/xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesEndpointFetchingService.java @@ -325,6 +325,9 @@ private void pushK8sEndpoints() { logger.debug("Pushing k8s endpoints: {}, group: {}", aggregator.getClusterName(), groupName); final ClusterLoadAssignment.Builder clusterLoadAssignmentBuilder = ClusterLoadAssignment.newBuilder().setClusterName(aggregator.getClusterName()); + if (aggregator.hasPolicy()) { + clusterLoadAssignmentBuilder.setPolicy(aggregator.getPolicy()); + } for (int i = 0; i < kubernetesEndpointGroupFutures.size(); i++) { final CompletableFuture future = diff --git a/xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesService.java b/xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesService.java index 3b3d9937f..30e1e4660 100644 --- a/xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesService.java +++ b/xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesService.java @@ -51,6 +51,7 @@ import com.linecorp.armeria.server.ServiceRequestContext; import com.linecorp.armeria.server.annotation.Blocking; import com.linecorp.armeria.server.annotation.Consumes; +import com.linecorp.armeria.server.annotation.Default; import com.linecorp.armeria.server.annotation.Delete; import com.linecorp.armeria.server.annotation.Param; import com.linecorp.armeria.server.annotation.Post; @@ -58,12 +59,15 @@ import com.linecorp.centraldogma.common.Author; import com.linecorp.centraldogma.common.EntryNotFoundException; import com.linecorp.centraldogma.common.RepositoryRole; +import com.linecorp.centraldogma.common.Revision; import com.linecorp.centraldogma.server.internal.credential.AccessTokenCredential; import com.linecorp.centraldogma.server.storage.repository.MetaRepository; import com.linecorp.centraldogma.xds.internal.RequiresXdsGroupRole; import com.linecorp.centraldogma.xds.internal.XdsResourceManager; import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment; +import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment.Policy; +import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment.Policy.DropOverload; import io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints; import io.fabric8.kubernetes.client.Config; import io.fabric8.kubernetes.client.ConfigBuilder; @@ -161,16 +165,20 @@ public CompletableFuture createKubernetesEndpointAggregator( bodyToStore = XdsResourceManager.injectYamlField(bodyToStore, "clusterName", clusterName); final String finalBodyToStore = bodyToStore; return validateKubernetesEndpointAndPushHttp( - kubernetesLocalityLbEndpointsList, group, aggregatorFileName, + kubernetesLocalityLbEndpointsList, aggregator.hasPolicy() ? aggregator.getPolicy() : null, + group, aggregatorFileName, () -> xdsResourceManager.push(group, kubernetesEndpointName, aggregatorFileName, createSummary, author, true, finalBodyToStore)); } /** - * PUT /xds/groups/{group}/k8s/endpointAggregators/{aggregator_id} + * PUT /xds/groups/{group}/k8s/endpointAggregators/{aggregator_id}?revision={baseRevision} * - *

Updates an existing Kubernetes endpoint aggregator. + *

Updates an existing Kubernetes endpoint aggregator. {@code revision} is the absolute revision the + * client read the aggregator at; the update is rejected with {@code 409 Conflict} if the group + * repository has advanced since, so a stale save cannot silently roll back a concurrent change. It + * defaults to {@code -1} (HEAD), which always applies. */ @Blocking @Put("/xds/groups/{group}/k8s/endpointAggregators/{*aggregator_id}") @@ -180,6 +188,7 @@ public CompletableFuture updateKubernetesEndpointAggregator( @Param("group") String group, @Param("aggregator_id") String aggregatorId, @Param("summary") @Nullable String summary, + @Param("revision") @Default("-1") Revision baseRevision, String body) { final String aggregatorName = "groups/" + group + K8S_ENDPOINT_AGGREGATORS_DIRECTORY + aggregatorId; final Matcher matcher = K8S_ENDPOINT_AGGREGATORS_NAME_PATTERN.matcher(aggregatorName); @@ -214,9 +223,10 @@ public CompletableFuture updateKubernetesEndpointAggregator( bodyToStore = XdsResourceManager.injectYamlField(bodyToStore, "clusterName", clusterName); final String finalBodyToStore = bodyToStore; return validateKubernetesEndpointAndPushHttp( - kubernetesLocalityLbEndpointsList, group, fileName(group, aggregatorName), + kubernetesLocalityLbEndpointsList, aggregator.hasPolicy() ? aggregator.getPolicy() : null, + group, fileName(group, aggregatorName), () -> xdsResourceManager.update(group, aggregatorName, updateSummary, author, - finalBodyToStore)); + finalBodyToStore, baseRevision)); } /** @@ -288,6 +298,9 @@ public CompletableFuture previewKubernetesEndpointAggregator( if (!aggregator.getClusterName().isEmpty()) { cla.setClusterName(aggregator.getClusterName()); } + if (aggregator.hasPolicy()) { + cla.setPolicy(aggregator.getPolicy()); + } for (CompletableFuture future : futures) { cla.addEndpoints(future.join()); } @@ -302,7 +315,7 @@ public CompletableFuture previewKubernetesEndpointAggregator( private CompletableFuture validateKubernetesEndpointAndPushHttp( List kubernetesLocalityLbEndpointsList, - String group, String fileNameForLookup, + @Nullable Policy policy, String group, String fileNameForLookup, Supplier> onSuccess) { for (KubernetesLocalityLbEndpoints kubernetesLocalityLbEndpoints : kubernetesLocalityLbEndpointsList) { try { @@ -312,6 +325,14 @@ private CompletableFuture validateKubernetesEndpointAndPushHttp( XdsResourceManager.errorResponse(HttpStatus.BAD_REQUEST, e)); } } + if (policy != null) { + try { + validatePolicy(policy); + } catch (IllegalArgumentException e) { + return CompletableFuture.completedFuture( + XdsResourceManager.errorResponse(HttpStatus.BAD_REQUEST, e)); + } + } final ContextAwareBlockingTaskExecutor taskExecutor = ServiceRequestContext.current().blockingTaskExecutor(); @@ -441,6 +462,23 @@ private static LocalityLbEndpoints toLocalityLbEndpoints( return builder.build(); } + // The policy is copied into the generated ClusterLoadAssignment, so a value Envoy rejects would take + // the whole endpoint set down with it. + private static void validatePolicy(Policy policy) { + if (policy.getDropOverloadsCount() > 1) { + throw new IllegalArgumentException( + "at most one drop_overload is supported, but got: " + policy.getDropOverloadsCount()); + } + for (DropOverload dropOverload : policy.getDropOverloadsList()) { + if (dropOverload.getCategory().isEmpty()) { + throw new IllegalArgumentException("category must not be empty in drop_overloads"); + } + } + if (policy.hasOverprovisioningFactor() && policy.getOverprovisioningFactor().getValue() == 0) { + throw new IllegalArgumentException("overprovisioning_factor must be greater than 0"); + } + } + private static void validateMetadataMappings(ServiceEndpointWatcher watcher) { for (MetadataMapping mapping : watcher.getMetadataMappingList()) { if (mapping.getResourceType() == MetadataMapping.ResourceType.RESOURCE_TYPE_UNSPECIFIED) { diff --git a/xds/src/test/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesServiceTest.java b/xds/src/test/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesServiceTest.java index db7d40b8e..ad7f710fc 100644 --- a/xds/src/test/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesServiceTest.java +++ b/xds/src/test/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesServiceTest.java @@ -34,6 +34,7 @@ import java.util.Queue; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; @@ -46,6 +47,7 @@ import com.fasterxml.jackson.databind.JsonNode; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; +import com.google.protobuf.UInt32Value; import com.linecorp.armeria.client.WebClient; import com.linecorp.armeria.common.AggregatedHttpResponse; @@ -67,6 +69,8 @@ import com.linecorp.centraldogma.xds.internal.XdsTestUtil; import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment; +import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment.Policy; +import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment.Policy.DropOverload; import io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints; import io.fabric8.kubernetes.api.model.Container; import io.fabric8.kubernetes.api.model.ContainerBuilder; @@ -266,6 +270,164 @@ void createEndpointAggregatorsRequest(String credentialId) throws IOException { assertNoContent(deleteAggregator0(aggregator.getName())); } + @Test + void policyIsPropagatedToTheGeneratedEndpoints() throws IOException { + final String aggregatorId = "policy-propagation-test"; + final Policy policy = Policy.newBuilder() + .setOverprovisioningFactor(UInt32Value.of(200)) + .setWeightedPriorityHealth(true) + .build(); + final KubernetesEndpointAggregator aggregator = + aggregator(aggregatorId, "repo-credential").toBuilder().setPolicy(policy).build(); + assertOk(createAggregator(aggregator, aggregatorId)); + + final Repository fooGroup = dogma.projectManager().get(INTERNAL_PROJECT_XDS).repos().get("foo"); + await().pollInterval(100, TimeUnit.MILLISECONDS).untilAsserted(() -> { + final Entry endpoints = + fooGroup.getOrNull(Revision.HEAD, + Query.ofYaml(K8S_ENDPOINTS_DIRECTORY + aggregatorId + ".yaml")) + .join(); + assertThat(endpoints).isNotNull(); + final JsonNode generatedPolicy = endpoints.content().get("policy"); + assertThat(generatedPolicy).isNotNull(); + assertThat(generatedPolicy.get("overprovisioningFactor").asInt()).isEqualTo(200); + assertThat(generatedPolicy.get("weightedPriorityHealth").asBoolean()).isTrue(); + }); + + assertNoContent(deleteAggregator0(aggregator.getName())); + } + + @Test + void policyEnvoyWouldRejectIsRefused() throws IOException { + final String aggregatorId = "policy-validation-test"; + final DropOverload drop = DropOverload.newBuilder().setCategory("throttle").build(); + final KubernetesEndpointAggregator aggregator = + aggregator(aggregatorId, "repo-credential").toBuilder() + .setPolicy(Policy.newBuilder() + .addDropOverloads(drop) + .addDropOverloads(drop.toBuilder().setCategory("lb"))) + .build(); + AggregatedHttpResponse response = createAggregator(aggregator, aggregatorId); + assertThat(response.status()).isSameAs(HttpStatus.BAD_REQUEST); + assertThat(response.contentUtf8()).contains("at most one drop_overload"); + + final KubernetesEndpointAggregator zeroFactor = + aggregator(aggregatorId, "repo-credential").toBuilder() + .setPolicy(Policy.newBuilder().setOverprovisioningFactor(UInt32Value.of(0))) + .build(); + response = createAggregator(zeroFactor, aggregatorId); + assertThat(response.status()).isSameAs(HttpStatus.BAD_REQUEST); + assertThat(response.contentUtf8()).contains("overprovisioning_factor"); + } + + @Test + void previewIncludesThePolicyThatWouldBeStored() throws IOException { + final Policy policy = Policy.newBuilder() + .setOverprovisioningFactor(UInt32Value.of(200)) + .build(); + final KubernetesEndpointAggregator aggregator = + aggregator("preview-policy-test", "repo-credential").toBuilder().setPolicy(policy).build(); + final RequestHeaders headers = + RequestHeaders.builder(HttpMethod.POST, + "/api/v1/xds/groups/foo/k8s/endpointAggregators:preview") + .set(HttpHeaderNames.AUTHORIZATION, "Bearer anonymous") + .contentType(MediaType.parse("application/yaml")) + .build(); + final AggregatedHttpResponse response = + dogma.httpClient().blocking().execute(headers, XdsTestUtil.toYaml(aggregator)); + assertOk(response); + assertThat(response.contentUtf8()).contains("overprovisioningFactor: 200"); + } + + @Test + void updateAggregatorRejectsStaleRevision() throws IOException { + final String aggregatorId = "update-cas-test"; + final KubernetesEndpointAggregator aggregator = aggregator(aggregatorId, "repo-credential"); + assertOk(createAggregator(aggregator, aggregatorId)); + + final Repository fooGroup = dogma.projectManager().get(INTERNAL_PROJECT_XDS).repos().get("foo"); + // KubernetesEndpointsUpdater commits the generated endpoints shortly after the aggregator is + // created; wait for it so a background commit cannot advance the head under this test. + await().pollInterval(100, TimeUnit.MILLISECONDS).untilAsserted(() -> { + assertThat(fooGroup.find(Revision.HEAD, K8S_ENDPOINTS_DIRECTORY + aggregatorId + ".yaml") + .join()).isNotEmpty(); + }); + final String aggregatorFile = K8S_ENDPOINT_AGGREGATORS_DIRECTORY + aggregatorId + ".yaml"; + final Revision loadedRevision = fooGroup.normalizeNow(Revision.HEAD); + final JsonNode storedContent = fooGroup.get(Revision.HEAD, Query.ofYaml(aggregatorFile)) + .join().content(); + + final KubernetesEndpointAggregator updatedAggregator = + aggregator.toBuilder() + .setLocalityLbEndpoints( + 0, aggregator.getLocalityLbEndpoints(0).toBuilder() + .setWatcher(aggregator.getLocalityLbEndpoints(0).getWatcher() + .toBuilder() + .setDistinctEndpoint(true))) + .build(); + + // Any commit in the group repository makes the loaded revision stale: the update is rejected and + // the stored file is left untouched. + bumpGroupRepository(); + final AggregatedHttpResponse conflict = + updateAggregatorWithRevision(updatedAggregator, aggregatorId, loadedRevision.text()); + assertThat(conflict.status()).isSameAs(HttpStatus.CONFLICT); + assertThat(fooGroup.get(Revision.HEAD, Query.ofYaml(aggregatorFile)).join().content()) + .isEqualTo(storedContent); + + // Retrying at the current revision applies. + assertOk(updateAggregatorWithRevision(updatedAggregator, aggregatorId, + fooGroup.normalizeNow(Revision.HEAD).text())); + final Entry entry = fooGroup.get(Revision.HEAD, Query.ofYaml(aggregatorFile)).join(); + assertThat(entry.content().get("localityLbEndpoints").get(0).get("watcher") + .get("distinctEndpoint").asBoolean()).isTrue(); + + assertNoContent(deleteAggregator0(aggregator.getName())); + } + + @Test + void updateAggregatorWithInvalidRevision() throws IOException { + final String aggregatorId = "update-cas-invalid-revision"; + final KubernetesEndpointAggregator aggregator = aggregator(aggregatorId, "repo-credential"); + assertOk(createAggregator(aggregator, aggregatorId)); + + // Unparsable revision. + AggregatedHttpResponse response = + updateAggregatorWithRevision(aggregator, aggregatorId, "not-a-revision"); + assertThat(response.status()).isSameAs(HttpStatus.BAD_REQUEST); + + // Well-formed but nonexistent (future) revision. + response = updateAggregatorWithRevision(aggregator, aggregatorId, "999999"); + assertThat(response.status()).isSameAs(HttpStatus.BAD_REQUEST); + assertThat(response.contentUtf8()).contains("Invalid revision"); + + assertNoContent(deleteAggregator0(aggregator.getName())); + } + + private static AggregatedHttpResponse updateAggregatorWithRevision( + KubernetesEndpointAggregator aggregator, String aggregatorId, + String revision) throws IOException { + final String path = "/api/v1/xds/groups/foo/k8s/endpointAggregators/" + aggregatorId + + "?revision=" + revision; + final RequestHeaders headers = RequestHeaders.builder(HttpMethod.PUT, path) + .contentType(MediaType.parse("application/yaml")) + .set(HttpHeaderNames.AUTHORIZATION, "Bearer anonymous") + .build(); + return dogma.httpClient().blocking().execute(headers, XdsTestUtil.toYaml(aggregator)); + } + + // Simulates a concurrent writer editing the aggregator file itself (not through the aggregator API): + // appends a semantic change so the parsed content differs, and returns the stored content after it. + private static int bumpCounter; + + // Commits an unrelated file, so a save that conflicts on it proves the compare-and-swap is scoped to + // the group repository rather than to the aggregator file. + private static void bumpGroupRepository() { + dogma.client().forRepo(INTERNAL_PROJECT_XDS, "foo") + .commit("bump", Change.ofTextUpsert("/bump.txt", "bump-" + ++bumpCounter)) + .push().join(); + } + private static KubernetesEndpointAggregator aggregator(String aggregatorId, String credentialId) { return aggregator(aggregatorId, "nginx-service", credentialId); } From a5a7f8300f1c6e114931e623e9ee75487d5fc9d0 Mon Sep 17 00:00:00 2001 From: Ikhun Um Date: Fri, 7 Aug 2026 10:38:45 +0900 Subject: [PATCH 2/5] Edit a K8s aggregator without losing what the form does not show Motivation: The aggregator form modelled part of the schema and rebuilt the document from that model on save, so every field it had no editor for was erased: distinctEndpoint, the metadata mappings, and the policy. An operator who edited an aggregator through the console discarded them without being told; losing distinctEndpoint, for instance, brings duplicate endpoints back on the next rolling restart. Modifications: - The form holds the stored document itself, so a field it renders is the field that gets saved and a field it does not render cannot be silently rebuilt away. What is left is pruning empty values before the document is written, and two adapters the document shape cannot express as form fields: the additional-properties map, whose keys are data, and the source-key oneof. - Add the missing editors: distinct endpoint, metadata mappings, and the policy. Drop overloads stay read-only because Envoy applies at most one and rejects the endpoints outright when it sees more; a stored one is shown and saved back unchanged. The two policy fields the Armeria xDS client does not read are marked as Envoy-only. - Send the revision the form was loaded at, and surface the server's 409 as a prompt to reload. Result: Editing an aggregator through the console keeps every field it was stored with, and a save based on a stale read is refused instead of rolling back a concurrent change. --- .../features/xds/K8sAggregatorEditor.tsx | 659 ++++++++++++++---- webapp/src/dogma/features/xds/xdsApiSlice.ts | 11 +- .../features/xds/K8sAggregatorEditor.test.tsx | 175 +++++ .../dogma/features/xds/xdsApiSlice.test.ts | 49 ++ 4 files changed, 738 insertions(+), 156 deletions(-) create mode 100644 webapp/tests/dogma/features/xds/xdsApiSlice.test.ts diff --git a/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx b/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx index dd7aee6c4..f4d549bd0 100644 --- a/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx +++ b/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx @@ -21,6 +21,7 @@ import { BreadcrumbItem, BreadcrumbLink, Button, + Badge, Checkbox, Flex, FormControl, @@ -30,16 +31,29 @@ import { HStack, IconButton, Input, + Select as ChakraSelect, SimpleGrid, Spacer, Text, + Tooltip, useDisclosure, } from '@chakra-ui/react'; +import { FetchBaseQueryError } from '@reduxjs/toolkit/query'; import * as jsYaml from 'js-yaml'; import { default as RouteLink } from 'next/link'; import Router from 'next/router'; import { useEffect, useState } from 'react'; -import { Control, Controller, FieldErrors, useFieldArray, useForm, UseFormRegister } from 'react-hook-form'; +import { + Control, + Controller, + FieldErrors, + useFieldArray, + useForm, + UseFormRegister, + UseFormSetValue, + UseFormGetValues, + useWatch, +} from 'react-hook-form'; import { OptionBase, Select } from 'chakra-react-select'; import { AiOutlineClose, AiOutlineDelete, AiOutlineEdit, AiOutlineEye } from 'react-icons/ai'; import { FiSave } from 'react-icons/fi'; @@ -67,107 +81,187 @@ import { K8sAggregatorStatus } from 'dogma/features/xds/K8sAggregatorStatus'; // Dots are allowed (e.g. "my-service.v1"), but slashes are not. const AGGREGATOR_ID_PATTERN = /^[a-z](?:[a-z0-9_.-]*[a-z0-9])?$/; -interface PropertyForm { +// The form holds the aggregator document itself, so a field the form edits is the field that gets saved. +interface DropOverload { + category?: string; + dropPercentage?: { numerator?: number; denominator?: string }; +} + +// A map whose keys are user input cannot be form field names, so the rows live here and the form value +// stays the map itself. Rows are kept in local state to preserve order and blank rows while typing. +const KeyValueEditor = ({ + value, + onChange, + readOnly, +}: { + value?: Record; + onChange: (value: Record) => void; + readOnly: boolean; +}) => { + const [rows, setRows] = useState(() => + Object.entries(value ?? {}).map(([key, v]) => ({ key, value: String(v) })), + ); + const update = (next: PropertyRow[]) => { + setRows(next); + const map: Record = {}; + next.forEach((row) => { + if (row.key.trim()) { + map[row.key.trim()] = row.value; + } + }); + onChange(map); + }; + return ( + <> + {rows.map((row, rowIndex) => ( + + update(rows.map((r, i) => (i === rowIndex ? { ...r, key: e.target.value } : r)))} + /> + update(rows.map((r, i) => (i === rowIndex ? { ...r, value: e.target.value } : r)))} + /> + {!readOnly && ( + } + onClick={() => update(rows.filter((_, i) => i !== rowIndex))} + /> + )} + + ))} + {!readOnly && ( + + )} + + ); +}; + +interface PropertyRow { key: string; value: string; } +interface MappingForm { + resourceType?: string; + entryType?: string; + sourceKey?: string; + sourceKeyPrefix?: string; + metadataNamespace?: string; + metadataKey?: string; +} + interface WatcherForm { - serviceName: string; - portName: string; - controlPlaneUrl: string; - namespace: string; - credentialId: string; - trustCerts: boolean; - priority: string; - loadBalancingWeight: string; - region: string; - zone: string; - subZone: string; - additionalProperties: PropertyForm[]; + serviceName?: string; + portName?: string; + kubeconfig: { + controlPlaneUrl?: string; + namespace?: string; + credentialId?: string; + trustCerts?: boolean; + }; + distinctEndpoint?: boolean; + metadataMapping: MappingForm[]; + additionalProperties?: Record; +} + +interface LocalityLbEndpointsForm { + watcher: WatcherForm; + locality: { region?: string; zone?: string; subZone?: string }; + priority?: number; + loadBalancingWeight?: number; } interface FormData { aggregatorId: string; - watchers: WatcherForm[]; + localityLbEndpoints: LocalityLbEndpointsForm[]; + policy: { + overprovisioningFactor?: number; + weightedPriorityHealth?: boolean; + endpointStaleAfter?: string; + // Not editable here — shown read-only and saved back as it was read. + dropOverloads?: DropOverload[]; + }; + // The revision the form was loaded at. Sent with the update so the server rejects a stale save. + loadedRevision?: string; } -const emptyWatcher: WatcherForm = { - serviceName: '', - portName: '', - controlPlaneUrl: '', - namespace: '', - credentialId: '', - trustCerts: false, - priority: '', - loadBalancingWeight: '', - region: '', - zone: '', - subZone: '', - additionalProperties: [], +const emptyMapping: MappingForm = { resourceType: 'NODE', entryType: 'LABEL' }; + +const emptyWatcher: LocalityLbEndpointsForm = { + watcher: { serviceName: '', kubeconfig: {}, metadataMapping: [] }, + locality: {}, }; -// Parses a numeric form field, rejecting non-numeric input instead of silently serializing it as null -// (JSON.stringify(NaN) === 'null'). The thrown error is surfaced to the user by the submit handler. -function toFiniteNumber(value: string, label: string): number { - const num = Number(value); - if (!Number.isFinite(num)) { - throw new Error(`${label} must be a number, but was '${value}'.`); +const emptyPolicy: FormData['policy'] = {}; + +// Drops what the server would reject or store as noise: blank strings, NaN from a cleared number input, and +// objects or arrays left empty once their own members were dropped. +// eslint-disable-next-line @typescript-eslint/no-explicit-any +function pruneEmpty(value: any): any { + if (Array.isArray(value)) { + const items = value.map(pruneEmpty).filter((v) => v !== undefined); + return items.length > 0 ? items : undefined; + } + if (value && typeof value === 'object') { + const out: Record = {}; + for (const [key, raw] of Object.entries(value)) { + const pruned = pruneEmpty(raw); + if (pruned !== undefined) { + out[key] = pruned; + } + } + return Object.keys(out).length > 0 ? out : undefined; } - return num; + if (typeof value === 'string') { + return value.trim() === '' ? undefined : value.trim(); + } + if (value === null || value === undefined || (typeof value === 'number' && isNaN(value))) { + return undefined; + } + if (value === false) { + return undefined; + } + return value; } function buildBody(data: FormData, name?: string): string { - const localityLbEndpoints = data.watchers.map((w) => { - const kubeconfig: Record = { controlPlaneUrl: w.controlPlaneUrl.trim() }; - if (w.namespace.trim()) { - kubeconfig.namespace = w.namespace.trim(); - } - if (w.credentialId.trim()) { - kubeconfig.credentialId = w.credentialId.trim(); - } - if (w.trustCerts) { - kubeconfig.trustCerts = true; - } - const watcher: Record = { serviceName: w.serviceName.trim(), kubeconfig }; - if (w.portName.trim()) { - watcher.portName = w.portName.trim(); - } - const additionalProperties: Record = {}; - w.additionalProperties.forEach((p) => { - if (p.key.trim()) { - additionalProperties[p.key.trim()] = p.value; - } - }); - if (Object.keys(additionalProperties).length > 0) { - watcher.additionalProperties = additionalProperties; - } - const entry: Record = { watcher }; - const locality: Record = {}; - if (w.region.trim()) { - locality.region = w.region.trim(); - } - if (w.zone.trim()) { - locality.zone = w.zone.trim(); - } - if (w.subZone.trim()) { - locality.subZone = w.subZone.trim(); - } - if (Object.keys(locality).length > 0) { - entry.locality = locality; - } - if (w.priority.trim()) { - entry.priority = toFiniteNumber(w.priority, 'Priority'); - } - if (w.loadBalancingWeight.trim()) { - entry.loadBalancingWeight = toFiniteNumber(w.loadBalancingWeight, 'Load balancing weight'); + const { dropOverloads, ...policy } = data.policy; + const pruned = pruneEmpty({ localityLbEndpoints: data.localityLbEndpoints, policy, name }); + const body = (pruned ?? {}) as { + localityLbEndpoints?: Record[]; + policy?: Record; + }; + if (dropOverloads && dropOverloads.length > 0) { + body.policy = { ...(body.policy ?? {}), dropOverloads }; + } + // Re-attached after pruning: a property is identified by its key, so an entry whose value is empty is a + // value the user chose, not an empty field to drop. + data.localityLbEndpoints.forEach((entry, index) => { + const additionalProperties = entry.watcher.additionalProperties; + const target = body.localityLbEndpoints?.[index]; + if (target && additionalProperties && Object.keys(additionalProperties).length > 0) { + target.watcher = { ...(target.watcher ?? {}), additionalProperties }; } - return entry; }); - const body: Record = { localityLbEndpoints }; - if (name) { - body.name = name; - } return jsYaml.dump(body); } @@ -177,30 +271,30 @@ function parseToFormData(aggregatorId: string, raw: any): FormData { // Throws YAMLException if raw is a string that is not valid YAML; callers must catch and notify the user. // eslint-disable-next-line @typescript-eslint/no-explicit-any const content: any = typeof raw === 'string' ? jsYaml.load(raw) : raw; - const endpoints = Array.isArray((content as any)?.localityLbEndpoints) - ? (content as any).localityLbEndpoints - : []; - const watchers: WatcherForm[] = endpoints.map( + const entries = Array.isArray(content?.localityLbEndpoints) ? content.localityLbEndpoints : []; + const localityLbEndpoints: LocalityLbEndpointsForm[] = entries.map( // eslint-disable-next-line @typescript-eslint/no-explicit-any - (e: any) => ({ - serviceName: e?.watcher?.serviceName ?? '', - portName: e?.watcher?.portName ?? '', - controlPlaneUrl: e?.watcher?.kubeconfig?.controlPlaneUrl ?? '', - namespace: e?.watcher?.kubeconfig?.namespace ?? '', - credentialId: e?.watcher?.kubeconfig?.credentialId ?? '', - trustCerts: !!e?.watcher?.kubeconfig?.trustCerts, - priority: e?.priority != null ? String(e.priority) : '', - loadBalancingWeight: e?.loadBalancingWeight != null ? String(e.loadBalancingWeight) : '', - region: e?.locality?.region ?? '', - zone: e?.locality?.zone ?? '', - subZone: e?.locality?.subZone ?? '', - additionalProperties: Object.entries(e?.watcher?.additionalProperties ?? {}).map(([key, value]) => ({ - key, - value: String(value), - })), + (entry: any) => ({ + ...entry, + locality: entry?.locality ?? {}, + watcher: { + ...entry?.watcher, + kubeconfig: entry?.watcher?.kubeconfig ?? {}, + metadataMapping: (Array.isArray(entry?.watcher?.metadataMapping) + ? entry.watcher.metadataMapping + : [] + ).map( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (rule: any) => rule, + ), + }, }), ); - return { aggregatorId, watchers: watchers.length > 0 ? watchers : [{ ...emptyWatcher }] }; + return { + aggregatorId, + localityLbEndpoints: localityLbEndpoints.length > 0 ? localityLbEndpoints : [{ ...emptyWatcher }], + policy: { ...emptyPolicy, ...content?.policy }, + }; } interface CredentialOption extends OptionBase { @@ -208,12 +302,139 @@ interface CredentialOption extends OptionBase { label: string; } +// One metadata mapping. The schema stores either sourceKey or sourceKeyPrefix, so the row offers one input +// and a selector; the selector is local state and switching it clears the field it turns off, which keeps the +// form value equal to the document. +const MappingRow = ({ + watcherIndex, + mappingIndex, + defaultValue, + register, + setValue, + getValues, + readOnly, + onRemove, +}: { + watcherIndex: number; + mappingIndex: number; + defaultValue: MappingForm; + register: UseFormRegister; + setValue: UseFormSetValue; + getValues: UseFormGetValues; + readOnly: boolean; + onRemove: () => void; +}) => { + const [prefixMode, setPrefixMode] = useState(defaultValue.sourceKeyPrefix != null); + const path = `localityLbEndpoints.${watcherIndex}.watcher.metadataMapping.${mappingIndex}` as const; + return ( + + + + Mapping #{mappingIndex + 1} + + + {!readOnly && ( + } + onClick={onRemove} + /> + )} + + + + Read from + + + + + + + Entry type + + + + + + + Match + { + const prefix = e.target.value === 'prefix'; + setPrefixMode(prefix); + // The document stores one key or the other, so move what was typed instead of dropping it. + if (prefix) { + setValue(`${path}.sourceKeyPrefix`, getValues(`${path}.sourceKey`) ?? ''); + setValue(`${path}.sourceKey`, ''); + // The server keeps the source keys in prefix mode and ignores this one. + setValue(`${path}.metadataKey`, ''); + } else { + setValue(`${path}.sourceKey`, getValues(`${path}.sourceKeyPrefix`) ?? ''); + setValue(`${path}.sourceKeyPrefix`, ''); + } + }} + > + + + + + + {prefixMode ? 'Source key prefix' : 'Source key'} + + + + Metadata namespace + + + + Metadata key + + + + + ); +}; + interface WatcherFieldsProps { index: number; control: Control; register: UseFormRegister; serviceNameError: boolean; controlPlaneUrlError: boolean; + setValue: UseFormSetValue; + getValues: UseFormGetValues; // The group's access-token credential ids to choose from, or null when they cannot be listed // (e.g. the user lacks the ADMIN role required by the credential API) — in which case a free-text input is // shown so the id can still be entered. @@ -229,14 +450,16 @@ const WatcherFields = ({ register, serviceNameError, controlPlaneUrlError, + setValue, + getValues, credentialOptions, onRemove, canRemove, readOnly, }: WatcherFieldsProps) => { - const { fields, append, remove } = useFieldArray({ + const mappings = useFieldArray({ control, - name: `watchers.${index}.additionalProperties` as `watchers.${number}.additionalProperties`, + name: `localityLbEndpoints.${index}.watcher.metadataMapping`, }); return ( @@ -256,7 +479,7 @@ const WatcherFields = ({ size="sm" placeholder="k8s service name" isReadOnly={readOnly} - {...register(`watchers.${index}.serviceName`, { required: true })} + {...register(`localityLbEndpoints.${index}.watcher.serviceName`, { required: true })} /> Service name is required. @@ -266,7 +489,7 @@ const WatcherFields = ({ size="sm" placeholder="optional" isReadOnly={readOnly} - {...register(`watchers.${index}.portName`)} + {...register(`localityLbEndpoints.${index}.watcher.portName`)} /> @@ -275,7 +498,7 @@ const WatcherFields = ({ size="sm" placeholder="https://kubernetes.default.svc" isReadOnly={readOnly} - {...register(`watchers.${index}.controlPlaneUrl`, { required: true })} + {...register(`localityLbEndpoints.${index}.watcher.kubeconfig.controlPlaneUrl`, { required: true })} /> Control plane URL is required. @@ -285,7 +508,7 @@ const WatcherFields = ({ size="sm" placeholder="optional" isReadOnly={readOnly} - {...register(`watchers.${index}.namespace`)} + {...register(`localityLbEndpoints.${index}.watcher.kubeconfig.namespace`)} /> @@ -293,7 +516,7 @@ const WatcherFields = ({ {credentialOptions !== null ? ( { const ids = value ? [...new Set([...credentialOptions, value])] : credentialOptions; const options: CredentialOption[] = ids.map((id) => ({ value: id, label: id })); @@ -320,7 +543,7 @@ const WatcherFields = ({ size="sm" placeholder="optional" isReadOnly={readOnly} - {...register(`watchers.${index}.credentialId`)} + {...register(`localityLbEndpoints.${index}.watcher.kubeconfig.credentialId`)} /> )} @@ -331,7 +554,7 @@ const WatcherFields = ({ type="number" placeholder="optional" isReadOnly={readOnly} - {...register(`watchers.${index}.priority`)} + {...register(`localityLbEndpoints.${index}.priority`, { valueAsNumber: true })} /> @@ -341,14 +564,25 @@ const WatcherFields = ({ type="number" placeholder="optional" isReadOnly={readOnly} - {...register(`watchers.${index}.loadBalancingWeight`)} + {...register(`localityLbEndpoints.${index}.loadBalancingWeight`, { valueAsNumber: true })} /> - + Trust certificates + + + Distinct endpoint + + @@ -361,7 +595,7 @@ const WatcherFields = ({ size="sm" placeholder="optional" isReadOnly={readOnly} - {...register(`watchers.${index}.region`)} + {...register(`localityLbEndpoints.${index}.locality.region`)} /> @@ -370,7 +604,7 @@ const WatcherFields = ({ size="sm" placeholder="optional" isReadOnly={readOnly} - {...register(`watchers.${index}.zone`)} + {...register(`localityLbEndpoints.${index}.locality.zone`)} /> @@ -379,7 +613,7 @@ const WatcherFields = ({ size="sm" placeholder="optional" isReadOnly={readOnly} - {...register(`watchers.${index}.subZone`)} + {...register(`localityLbEndpoints.${index}.locality.subZone`)} /> @@ -387,40 +621,38 @@ const WatcherFields = ({ Additional properties (optional) - {fields.map((field, propIndex) => ( - - - - {!readOnly && ( - } - onClick={() => remove(propIndex)} - /> - )} - + ( + + )} + /> + + + Metadata mappings (optional) + + {mappings.fields.map((field, mappingIndex) => ( + mappings.remove(mappingIndex)} + /> ))} {!readOnly && ( )} @@ -431,6 +663,8 @@ interface AggregatorFormFieldsProps { group: string; control: Control; register: UseFormRegister; + setValue: UseFormSetValue; + getValues: UseFormGetValues; errors: FieldErrors; idReadOnly: boolean; readOnly: boolean; @@ -440,11 +674,13 @@ const AggregatorFormFields = ({ group, control, register, + setValue, + getValues, errors, idReadOnly, readOnly, }: AggregatorFormFieldsProps) => { - const { fields, append, remove } = useFieldArray({ control, name: 'watchers' }); + const { fields, append, remove } = useFieldArray({ control, name: 'localityLbEndpoints' }); // Offer the group's access-token credentials as a dropdown. Listing requires the ADMIN role, so on any error // (e.g. 403 for non-admins) fall back to a free-text credential id input. const { data: credentials, error: credentialsError } = useListCredentialsQuery({ group }); @@ -476,8 +712,10 @@ const AggregatorFormFields = ({ index={index} control={control} register={register} - serviceNameError={!!errors.watchers?.[index]?.serviceName} - controlPlaneUrlError={!!errors.watchers?.[index]?.controlPlaneUrl} + setValue={setValue} + getValues={getValues} + serviceNameError={!!errors.localityLbEndpoints?.[index]?.watcher?.serviceName} + controlPlaneUrlError={!!errors.localityLbEndpoints?.[index]?.watcher?.kubeconfig?.controlPlaneUrl} credentialOptions={credentialOptions} onRemove={() => remove(index)} canRemove={fields.length > 1} @@ -494,10 +732,100 @@ const AggregatorFormFields = ({ Add watcher )} + + + {/* Room for the sticky action bar, which would otherwise cover the last fields. */} + ); }; +// A drop percentage is a numerator over a chosen denominator; render it the way an operator reads it. +function formatDropShare(drop: DropOverload): string { + const numerator = drop.dropPercentage?.numerator ?? 0; + const denominator = drop.dropPercentage?.denominator ?? 'HUNDRED'; + if (denominator === 'HUNDRED') { + return `${numerator}%`; + } + return `${numerator} in ${(denominator === 'MILLION' ? 1_000_000 : 10_000).toLocaleString('en-US')}`; +} + +// Marks a policy field the Armeria xDS client does not read, so an operator does not expect an effect. +const EnvoyOnlyBadge = () => ( + + + Envoy only + + +); + +// Load-balancing policy of the generated ClusterLoadAssignment. Envoy honours every field the form offers; +// the Armeria xDS client reads only the first two, so the rest is marked to set the operator's expectation. +const PolicyFields = ({ + control, + register, + readOnly, +}: { + control: Control; + register: UseFormRegister; + readOnly: boolean; +}) => { + const dropOverloads = useWatch({ control, name: 'policy.dropOverloads' }) as DropOverload[] | undefined; + return ( + + + Policy (optional) + + + + Overprovisioning factor + + + + + Weighted priority health + + + + + + + Endpoint stale after + + + + + {dropOverloads && dropOverloads.length > 0 && ( + + + Drop overload + + + + Envoy drops this share of requests to the cluster. Edit it where it was set. + + {dropOverloads.map((drop, dropIndex) => ( + + {drop.category ?? '(no category)'} — {formatDropShare(drop)} + + ))} + + )} + + ); +}; + const NewK8sAggregatorEditor = ({ group }: { group: string }) => { const dispatch = useAppDispatch(); // Creating an aggregator requires WRITE on the group, mirroring the Edit/Delete gating in @@ -511,9 +839,13 @@ const NewK8sAggregatorEditor = ({ group }: { group: string }) => { const { register, control, + setValue, + getValues, handleSubmit, formState: { errors }, - } = useForm({ defaultValues: { aggregatorId: '', watchers: [{ ...emptyWatcher }] } }); + } = useForm({ + defaultValues: { aggregatorId: '', localityLbEndpoints: [{ ...emptyWatcher }], policy: { ...emptyPolicy } }, + }); const onPreview = async (data: FormData) => { setPreviewResult(null); @@ -564,6 +896,8 @@ const NewK8sAggregatorEditor = ({ group }: { group: string }) => { group={group} control={control} register={register} + setValue={setValue} + getValues={getValues} errors={errors} idReadOnly={false} readOnly={false} @@ -618,17 +952,22 @@ const ExistingK8sAggregatorEditor = ({ group, id }: { group: string; id: string const { register, control, + setValue, + getValues, handleSubmit, reset, formState: { errors }, - } = useForm({ defaultValues: { aggregatorId: id, watchers: [{ ...emptyWatcher }] } }); + } = useForm({ + defaultValues: { aggregatorId: id, localityLbEndpoints: [{ ...emptyWatcher }], policy: { ...emptyPolicy } }, + }); // Sync the form to the latest fetched content, but never while editing so a background refetch cannot // clobber unsaved edits. useEffect(() => { if (data && !editing) { try { - reset(parseToFormData(id, (data as FileContentDto).content)); + const file = data as FileContentDto; + reset({ ...parseToFormData(id, file.content), loadedRevision: String(file.revision) }); } catch (e) { dispatch(newNotification('Failed to load aggregator', (e as Error).message, 'error')); } @@ -643,19 +982,31 @@ const ExistingK8sAggregatorEditor = ({ group, id }: { group: string; id: string id, body: buildBody(formData, name), summary: commitSummary || undefined, + revision: String(formData.loadedRevision), }).unwrap(); dispatch(newNotification('Aggregator updated', `Aggregator '${id}' is updated`, 'success')); setEditing(false); setCommitSummary(''); } catch (err) { - dispatch(newNotification('Failed to update the aggregator', ErrorMessageParser.parse(err), 'error')); + if ((err as FetchBaseQueryError | undefined)?.status === 409) { + dispatch( + newNotification( + 'Update conflict', + `Group '${group}' changed after you loaded this aggregator. Reload the page and re-apply your edits.`, + 'error', + ), + ); + } else { + dispatch(newNotification('Failed to update the aggregator', ErrorMessageParser.parse(err), 'error')); + } } }; const handleCancel = () => { if (data) { try { - reset(parseToFormData(id, (data as FileContentDto).content)); + const file = data as FileContentDto; + reset({ ...parseToFormData(id, file.content), loadedRevision: String(file.revision) }); } catch (e) { dispatch(newNotification('Failed to restore aggregator content', (e as Error).message, 'error')); } @@ -728,6 +1079,8 @@ const ExistingK8sAggregatorEditor = ({ group, id }: { group: string; id: string group={group} control={control} register={register} + setValue={setValue} + getValues={getValues} errors={errors} idReadOnly readOnly={!editing} diff --git a/webapp/src/dogma/features/xds/xdsApiSlice.ts b/webapp/src/dogma/features/xds/xdsApiSlice.ts index 18c0d8b65..9dc84a546 100644 --- a/webapp/src/dogma/features/xds/xdsApiSlice.ts +++ b/webapp/src/dogma/features/xds/xdsApiSlice.ts @@ -295,11 +295,16 @@ export const xdsApiSlice = createApi({ }), updateK8sAggregator: builder.mutation< unknown, - { group: string; id: string; body: string; summary?: string } + // `revision` is the revision the client read the aggregator at; the server rejects the update with + // 409 when the aggregator changed since. Required so no caller opts out of the check. + { group: string; id: string; body: string; summary?: string; revision: string } >({ - query: ({ group, id, body, summary }) => { + query: ({ group, id, body, summary, revision }) => { let url = `/api/v1/xds/groups/${group}/k8s/endpointAggregators/${id}`; - if (summary) url += `?summary=${encodeURIComponent(summary)}`; + const params = new URLSearchParams(); + if (summary) params.set('summary', summary); + params.set('revision', revision); + url += `?${params.toString()}`; return { url, method: 'PUT', body, headers: { 'Content-Type': 'application/yaml' } }; }, invalidatesTags: ['K8sAggregator'], diff --git a/webapp/tests/dogma/features/xds/K8sAggregatorEditor.test.tsx b/webapp/tests/dogma/features/xds/K8sAggregatorEditor.test.tsx index 36a35778a..b8bff471f 100644 --- a/webapp/tests/dogma/features/xds/K8sAggregatorEditor.test.tsx +++ b/webapp/tests/dogma/features/xds/K8sAggregatorEditor.test.tsx @@ -165,6 +165,181 @@ describe('K8sAggregatorEditor – aggregator ID pattern validation', () => { }); }); + describe('fields the form must not drop', () => { + it('round-trips distinctEndpoint, metadataMapping and policy, and sends the loaded revision', async () => { + const stored = { + localityLbEndpoints: [ + { + watcher: { + serviceName: 'my-service', + kubeconfig: { controlPlaneUrl: 'https://kubernetes.default.svc' }, + distinctEndpoint: true, + metadataMapping: [ + { resourceType: 'NODE', entryType: 'LABEL', sourceKey: 'topology.kubernetes.io/zone' }, + ], + }, + }, + ], + policy: { overprovisioningFactor: 200, weightedPriorityHealth: true }, + }; + jest.mocked(xdsApiSlice.useGetK8sAggregatorQuery).mockReturnValue({ + data: { content: jsYaml.dump(stored), revision: 7 }, + isLoading: false, + error: undefined, + } as any); + + const user = userEvent.setup(); + renderWithProviders(); + await waitFor(() => expect(screen.getByDisplayValue('my-agg')).toBeInTheDocument()); + + // Every stored field is on screen, not carried invisibly. + expect(screen.getByLabelText(/distinct endpoint/i)).toBeChecked(); + expect(screen.getByDisplayValue('topology.kubernetes.io/zone')).toBeInTheDocument(); + expect(screen.getByDisplayValue('200')).toBeInTheDocument(); + expect(screen.getByLabelText(/weighted priority health/i)).toBeChecked(); + + await user.click(screen.getByRole('button', { name: /^edit$/i })); + await user.click(screen.getByRole('button', { name: /^save$/i })); + + await waitFor(() => expect(mockUpdate).toHaveBeenCalled()); + const sent = jsYaml.load(mockUpdate.mock.calls[0][0].body) as any; + const watcher = sent.localityLbEndpoints[0].watcher; + expect(watcher.distinctEndpoint).toBe(true); + expect(watcher.metadataMapping).toEqual([ + { resourceType: 'NODE', entryType: 'LABEL', sourceKey: 'topology.kubernetes.io/zone' }, + ]); + expect(sent.policy).toEqual({ overprovisioningFactor: 200, weightedPriorityHealth: true }); + // The revision the form was loaded at rides with the update so the server can reject a stale save. + expect(mockUpdate.mock.calls[0][0].revision).toBe('7'); + }); + + it('shows a stored drop overload read-only and saves it back unchanged', async () => { + const stored = { + localityLbEndpoints: [ + { + watcher: { + serviceName: 'my-service', + kubeconfig: { controlPlaneUrl: 'https://kubernetes.default.svc' }, + }, + }, + ], + policy: { dropOverloads: [{ category: 'throttle', dropPercentage: { numerator: 30 } }] }, + }; + jest.mocked(xdsApiSlice.useGetK8sAggregatorQuery).mockReturnValue({ + data: { content: jsYaml.dump(stored), revision: 7 }, + isLoading: false, + error: undefined, + } as any); + + const user = userEvent.setup(); + renderWithProviders(); + await waitFor(() => expect(screen.getByDisplayValue('my-agg')).toBeInTheDocument()); + + // Visible, but with no input to change it. + expect(screen.getByText(/throttle — 30%/)).toBeInTheDocument(); + + await user.click(screen.getByRole('button', { name: /^edit$/i })); + await user.click(screen.getByRole('button', { name: /^save$/i })); + + await waitFor(() => expect(mockUpdate).toHaveBeenCalled()); + const sent = jsYaml.load(mockUpdate.mock.calls[0][0].body) as any; + expect(sent.policy.dropOverloads).toEqual([{ category: 'throttle', dropPercentage: { numerator: 30 } }]); + }); + + it('round-trips every field the form renders', async () => { + // The refactor rewrote each register() path, and a wrong path drops that field silently. + const stored = { + localityLbEndpoints: [ + { + watcher: { + serviceName: 'my-service', + portName: 'http', + kubeconfig: { + controlPlaneUrl: 'https://kubernetes.default.svc', + namespace: 'prod', + credentialId: 'my-credential', + trustCerts: true, + }, + metadataMapping: [ + { + resourceType: 'POD', + entryType: 'ANNOTATION', + sourceKeyPrefix: 'topology.kubernetes.io/', + metadataNamespace: 'envoy.lb', + }, + ], + }, + locality: { region: 'us-east-1', zone: 'us-east-1a', subZone: 'rack-3' }, + priority: 1, + loadBalancingWeight: 50, + }, + ], + policy: { endpointStaleAfter: '30s' }, + }; + jest.mocked(xdsApiSlice.useGetK8sAggregatorQuery).mockReturnValue({ + data: { content: jsYaml.dump(stored), revision: 7 }, + isLoading: false, + error: undefined, + } as any); + + const user = userEvent.setup(); + renderWithProviders(); + await waitFor(() => expect(screen.getByDisplayValue('my-agg')).toBeInTheDocument()); + + await user.click(screen.getByRole('button', { name: /^edit$/i })); + await user.click(screen.getByRole('button', { name: /^save$/i })); + + await waitFor(() => expect(mockUpdate).toHaveBeenCalled()); + const sent = jsYaml.load(mockUpdate.mock.calls[0][0].body) as any; + expect(sent.localityLbEndpoints).toEqual(stored.localityLbEndpoints); + expect(sent.policy).toEqual(stored.policy); + }); + + it('keeps an additional property whose value is empty', async () => { + const stored = { + localityLbEndpoints: [ + { + watcher: { + serviceName: 'my-service', + kubeconfig: { controlPlaneUrl: 'https://kubernetes.default.svc' }, + // An empty label value is valid in Kubernetes, so it must survive a save. + additionalProperties: { nodeIpLabel: '' }, + }, + }, + ], + }; + jest.mocked(xdsApiSlice.useGetK8sAggregatorQuery).mockReturnValue({ + data: { content: jsYaml.dump(stored), revision: 7 }, + isLoading: false, + error: undefined, + } as any); + + const user = userEvent.setup(); + renderWithProviders(); + await waitFor(() => expect(screen.getByDisplayValue('my-agg')).toBeInTheDocument()); + + await user.click(screen.getByRole('button', { name: /^edit$/i })); + await user.click(screen.getByRole('button', { name: /^save$/i })); + + await waitFor(() => expect(mockUpdate).toHaveBeenCalled()); + const sent = jsYaml.load(mockUpdate.mock.calls[0][0].body) as any; + expect(sent.localityLbEndpoints[0].watcher.additionalProperties).toEqual({ nodeIpLabel: '' }); + }); + + it('surfaces a 409 as an update conflict', async () => { + mockUpdate.mockReturnValue({ unwrap: () => Promise.reject({ status: 409 }) }); + const user = userEvent.setup(); + const { store } = renderWithProviders(); + await waitFor(() => expect(screen.getByDisplayValue('my-agg')).toBeInTheDocument()); + + await user.click(screen.getByRole('button', { name: /^edit$/i })); + await user.click(screen.getByRole('button', { name: /^save$/i })); + + await waitFor(() => expect(store.getState().notification.title).toBe('Update conflict')); + expect(screen.getByRole('button', { name: /^save$/i })).toBeInTheDocument(); + }); + }); + describe('sticky action bar', () => { it('moves Cancel into the bar and reveals the commit input + Save only while editing', async () => { const user = userEvent.setup(); diff --git a/webapp/tests/dogma/features/xds/xdsApiSlice.test.ts b/webapp/tests/dogma/features/xds/xdsApiSlice.test.ts new file mode 100644 index 000000000..a8c0418b3 --- /dev/null +++ b/webapp/tests/dogma/features/xds/xdsApiSlice.test.ts @@ -0,0 +1,49 @@ +/* + * Copyright 2026 LY Corporation + * + * LY Corporation licenses this file to you 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: + * + * https://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. + */ +// The component tests mock the hooks, so this drives the real endpoint with a stubbed fetch. The revision +// parameter name has to match the server's @Param("revision"), or a stale save silently applies. +import 'whatwg-fetch'; +import { setupStore } from 'dogma/store'; +import { xdsApiSlice } from 'dogma/features/xds/xdsApiSlice'; + +describe('xdsApiSlice – updateK8sAggregator', () => { + afterEach(() => jest.restoreAllMocks()); + + it('sends a PUT carrying the summary and the loaded revision', async () => { + const fetchSpy = jest + .spyOn(window, 'fetch') + .mockResolvedValue( + new Response('stored: yaml\n', { status: 200, headers: { 'Content-Type': 'application/yaml' } }), + ); + + await setupStore().dispatch( + xdsApiSlice.endpoints.updateK8sAggregator.initiate({ + group: 'foo', + id: 'my-agg', + body: 'a: b\n', + summary: 'update & verify', + revision: '7', + }), + ); + + const request = fetchSpy.mock.calls[0][0] as Request; + expect(request.method).toBe('PUT'); + const url = new URL(request.url, 'http://localhost'); + expect(url.pathname).toBe('/api/v1/xds/groups/foo/k8s/endpointAggregators/my-agg'); + expect(url.searchParams.get('summary')).toBe('update & verify'); + expect(url.searchParams.get('revision')).toBe('7'); + }); +}); From 0f3105635592502206e71f19271682b036376c12 Mon Sep 17 00:00:00 2001 From: Ikhun Um Date: Mon, 10 Aug 2026 15:50:33 +0900 Subject: [PATCH 3/5] Keep the dev server's build directory out of the gradle build's way MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Motivation: `next dev` and `npm run build` both write to `build/web`, and `:webapp:runTestServer` runs the latter. Starting a test server while a dev server is up therefore deletes the dev server's manifests, and the page turns into "missing required error components, refreshing...", a 500, or silently stops hot-reloading — with nothing to suggest the backend did it. Modifications: - `distDir` reads NEXT_DIST_DIR when set, so the dev server can be pointed elsewhere. The default is unchanged, so the production build and CI are unaffected. Result: `NEXT_DIST_DIR=.next npm run develop` survives a gradle build running beside it. --- webapp/next.config.js | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/webapp/next.config.js b/webapp/next.config.js index 06af16e3a..4654bfc18 100644 --- a/webapp/next.config.js +++ b/webapp/next.config.js @@ -11,7 +11,9 @@ const nextConfig = { productionBrowserSourceMaps: isDev, trailingSlash: false, output: isDev ? 'standalone' : 'export', - distDir: 'build/web/', + // `npm run build` and `next dev` share this directory, so a gradle build (which runs the former as part + // of :webapp:runTestServer) wipes a running dev server's manifests. Point the dev server elsewhere. + distDir: process.env.NEXT_DIST_DIR || 'build/web/', images: { unoptimized: true, }, From 59069eff42f5f09eb98c539525100ed17b47344f Mon Sep 17 00:00:00 2001 From: Ikhun Um Date: Mon, 10 Aug 2026 15:50:51 +0900 Subject: [PATCH 4/5] Explain the K8s aggregator form to someone seeing it for the first time Motivation: The form named the schema's fields and left it at that. An operator who had not written this YAML before could not tell what a watcher was, which direction "Trust certificates" was safe in, that priority 0 is the highest, or what a metadata mapping copies from where. The schema was on screen; its meaning was not. Modifications: - Every field carries a one-line hint, and each group carries a caption: what reaching the cluster needs, what ends up in the endpoints, what a locality is reported for, what a mapping copies. - Split each source into "Cluster access" and "Endpoints", so how to reach Kubernetes is separate from what is read out of it, and name the card after what it is: a Kubernetes endpoint source. - Give the page a type scale rather than three weights of the same size: sections at 12px uppercase on a rule, labels at 14px semibold, hints at 12px. Policy is one field per row, and mapping rows pair the six fields two by two, so nothing wraps or sits alone. - A read-only select no longer dims the value it is showing, which made a chosen value look unset. Result: The form reads as an explanation of what it is about to save, not as a list of proto field names. --- .../features/xds/K8sAggregatorEditor.tsx | 211 ++++++++++++------ 1 file changed, 137 insertions(+), 74 deletions(-) diff --git a/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx b/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx index f4d549bd0..ccc8be0fb 100644 --- a/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx +++ b/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx @@ -23,10 +23,14 @@ import { Button, Badge, Checkbox, + Divider, Flex, FormControl, FormErrorMessage, + FormHelperText, + FormHelperTextProps, FormLabel, + FormLabelProps, Heading, HStack, IconButton, @@ -34,15 +38,17 @@ import { Select as ChakraSelect, SimpleGrid, Spacer, + Stack, Text, Tooltip, + useColorModeValue, useDisclosure, } from '@chakra-ui/react'; import { FetchBaseQueryError } from '@reduxjs/toolkit/query'; import * as jsYaml from 'js-yaml'; import { default as RouteLink } from 'next/link'; import Router from 'next/router'; -import { useEffect, useState } from 'react'; +import { ReactNode, useEffect, useState } from 'react'; import { Control, Controller, @@ -87,6 +93,26 @@ interface DropOverload { dropPercentage?: { numerator?: number; denominator?: string }; } +// A section label that reads as a boundary: the rule carries the eye across, the label sits on it. +const SectionLabel = ({ children, mt = 6 }: { children: ReactNode; mt?: number }) => ( + + + {children} + + + +); + +// Chakra's FormLabel is medium; semibold separates it from the hint underneath. +const Label = (props: FormLabelProps) => ; + +// Chakra's FormHelperText defaults to gray.500 / whiteAlpha.600, which is too faint to read in dark mode. +const Help = ({ children, ...props }: FormHelperTextProps) => ( + + {children} + +); + // A map whose keys are user input cannot be form field names, so the rows live here and the form value // stays the map itself. Rows are kept in local state to preserve order and blank rows while typing. const KeyValueEditor = ({ @@ -344,9 +370,9 @@ const MappingRow = ({ /> )} - + - Read from + Node + Read the value from the Pod or its Node. - Entry type + Label + Read it from a label or an annotation. - Match + { const prefix = e.target.value === 'prefix'; @@ -393,27 +422,30 @@ const MappingRow = ({ + Copy one key, or every key with a prefix. - {prefixMode ? 'Source key prefix' : 'Source key'} + + {prefixMode ? 'Copy every key starting with this.' : 'The key to copy the value from.'} - Metadata namespace + + Stored under this namespace. Defaults to envoy.lb. - Metadata key + + + {prefixMode + ? 'Unused — the source keys are kept.' + : 'Stored under this key. Defaults to the source key.'} + @@ -464,7 +501,7 @@ const WatcherFields = ({ return ( - Watcher #{index + 1} + Kubernetes endpoint source #{index + 1} {canRemove && !readOnly && ( )} - - - Service name - - Service name is required. - - - Port name - - + Cluster access + - Control plane URL + Control plane URL is required. + The Kubernetes API server to read from. - Namespace + + Defaults to the credential's namespace. - Credential ID + {credentialOptions !== null ? ( )} + Empty if the cluster needs none. + + + + Trust certificates + + Skips TLS verification. Only for a self-signed control plane. + + + + Endpoints + + + + + Service name is required. + Its Pods become the endpoints. + + + + + Only when the Service has several ports. - Priority + + 0 is highest; the next takes over. - Load balancing weight + + Share of traffic relative to the other sources. - - - Trust certificates - - - + Distinct endpoint + Collapses endpoints sharing a host and port. - - Locality (optional) + Locality (optional) + + Reported to Envoy so it can prefer endpoints close to the caller. - + - Region + - Zone + - Sub zone + - - Additional properties (optional) + Additional properties (optional) + + Passed to the server-side resolvers. - - Metadata mappings (optional) + Metadata mappings (optional) + + Copies Pod or Node labels and annotations into the endpoint metadata for routing rules to match on. {mappings.fields.map((field, mappingIndex) => ( - Aggregator ID + ID must match [a-z](?:[a-z0-9_.-]*[a-z0-9])? (dots allowed, slashes not allowed) + Names the aggregator and its cluster. {fields.map((field, index) => ( @@ -729,7 +788,7 @@ const AggregatorFormFields = ({ leftIcon={} onClick={() => append({ ...emptyWatcher })} > - Add watcher + Add source )} @@ -776,36 +835,40 @@ const PolicyFields = ({ Policy (optional) - + - Overprovisioning factor + + Healthy above 100/factor — 140 means 72%. - - + + Weighted priority health + Weighs priority health by endpoint weight, not count. - - - - - Endpoint stale after - - - - + + + + Drops an endpoint unrefreshed for this long. + + {dropOverloads && dropOverloads.length > 0 && ( From 9f0f7991e2f9f8a4f3e46d54deeadb9417166a5a Mon Sep 17 00:00:00 2001 From: Ikhun Um Date: Tue, 11 Aug 2026 12:57:46 +0900 Subject: [PATCH 5/5] Make the aggregator form easier to work through Motivation: The form put everything on screen at once and left the reader to work out what belonged together. The policy, which most aggregators never set, took as much room as the fields that matter; the button that adds a source read as part of the card above it; and adding one left the cursor wherever it happened to be rather than in the new card. Modifications: - Collapse the policy behind its own header, with a line saying what it is for, and open it whenever one is stored. In read mode an aggregator without a policy no longer shows an empty section. - Give the sources their own colour, shared by the card heading and the button that adds one, so the two read as the same thing. The button sits between the cards and the policy, out of both. - Adding a source moves the cursor to the new card's first field. - Drop the hint under the aggregator ID, which said what the label already says. Result: The form opens on the fields an operator actually fills in, and adding a second source continues where the typing left off. --- .../features/xds/K8sAggregatorEditor.tsx | 67 +++++++++++++++---- 1 file changed, 53 insertions(+), 14 deletions(-) diff --git a/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx b/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx index ccc8be0fb..51d9716aa 100644 --- a/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx +++ b/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx @@ -32,6 +32,7 @@ import { FormLabel, FormLabelProps, Heading, + Icon, HStack, IconButton, Input, @@ -58,12 +59,14 @@ import { UseFormRegister, UseFormSetValue, UseFormGetValues, + UseFormSetFocus, useWatch, } from 'react-hook-form'; import { OptionBase, Select } from 'chakra-react-select'; import { AiOutlineClose, AiOutlineDelete, AiOutlineEdit, AiOutlineEye } from 'react-icons/ai'; import { FiSave } from 'react-icons/fi'; import { IoAddCircleOutline } from 'react-icons/io5'; +import { MdExpandLess, MdExpandMore } from 'react-icons/md'; import { Deferred } from 'dogma/common/components/Deferred'; import { DeleteConfirmationModal } from 'dogma/common/components/DeleteConfirmationModal'; import { @@ -501,7 +504,9 @@ const WatcherFields = ({ return ( - Kubernetes endpoint source #{index + 1} + + Kubernetes endpoint source #{index + 1} + {canRemove && !readOnly && ( + + + )} @@ -829,13 +841,36 @@ const PolicyFields = ({ register: UseFormRegister; readOnly: boolean; }) => { + // Watched leaf by leaf: watching the `policy` object itself does not re-render when reset() fills it in. + const overprovisioningFactor = useWatch({ control, name: 'policy.overprovisioningFactor' }); + const weightedPriorityHealth = useWatch({ control, name: 'policy.weightedPriorityHealth' }); + const endpointStaleAfter = useWatch({ control, name: 'policy.endpointStaleAfter' }); const dropOverloads = useWatch({ control, name: 'policy.dropOverloads' }) as DropOverload[] | undefined; + // Most aggregators set no policy, so the section opens only when one is stored or asked for. Derived + // rather than initialised, because the stored values arrive after this mounts. + const [opened, setOpened] = useState(false); + const stored = + overprovisioningFactor != null || + !!weightedPriorityHealth || + !!endpointStaleAfter || + !!dropOverloads?.length; + const expanded = opened || stored; + if (!expanded && readOnly) { + return null; + } return ( - - Policy (optional) - - + setOpened(!expanded)}> + + Policy (optional) + + {!expanded && ( + + How Envoy balances across these endpoints + + )} + + { control, setValue, getValues, + setFocus, handleSubmit, formState: { errors }, } = useForm({ @@ -961,6 +997,7 @@ const NewK8sAggregatorEditor = ({ group }: { group: string }) => { register={register} setValue={setValue} getValues={getValues} + setFocus={setFocus} errors={errors} idReadOnly={false} readOnly={false} @@ -1017,6 +1054,7 @@ const ExistingK8sAggregatorEditor = ({ group, id }: { group: string; id: string control, setValue, getValues, + setFocus, handleSubmit, reset, formState: { errors }, @@ -1144,6 +1182,7 @@ const ExistingK8sAggregatorEditor = ({ group, id }: { group: string; id: string register={register} setValue={setValue} getValues={getValues} + setFocus={setFocus} errors={errors} idReadOnly readOnly={!editing}