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, }, diff --git a/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx b/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx index dd7aee6c4..51d9716aa 100644 --- a/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx +++ b/webapp/src/dogma/features/xds/K8sAggregatorEditor.tsx @@ -21,29 +21,52 @@ import { BreadcrumbItem, BreadcrumbLink, Button, + Badge, Checkbox, + Divider, Flex, FormControl, FormErrorMessage, + FormHelperText, + FormHelperTextProps, FormLabel, + FormLabelProps, Heading, + Icon, HStack, IconButton, Input, + 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 { Control, Controller, FieldErrors, useFieldArray, useForm, UseFormRegister } from 'react-hook-form'; +import { ReactNode, useEffect, useState } from 'react'; +import { + Control, + Controller, + FieldErrors, + useFieldArray, + useForm, + 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 { @@ -67,107 +90,207 @@ 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 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 = ({ + 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; + } + if (typeof value === 'string') { + return value.trim() === '' ? undefined : value.trim(); + } + if (value === null || value === undefined || (typeof value === 'number' && isNaN(value))) { + return undefined; } - return num; + 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 +300,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 +331,150 @@ 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 the value from the Pod or its Node. + + + + + + + + Read it from a label or an annotation. + + + + { + 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`, ''); + } + }} + > + + + + Copy one key, or every key with a prefix. + + + + + {prefixMode ? 'Copy every key starting with this.' : 'The key to copy the value from.'} + + + + + Stored under this namespace. Defaults to envoy.lb. + + + + + + {prefixMode + ? 'Unused — the source keys are kept.' + : 'Stored under this key. Defaults to the source 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,19 +490,23 @@ 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 ( - 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 ? ( { const ids = value ? [...new Set([...credentialOptions, value])] : credentialOptions; const options: CredentialOption[] = ids.map((id) => ({ value: id, label: id })); @@ -320,107 +571,151 @@ const WatcherFields = ({ size="sm" placeholder="optional" isReadOnly={readOnly} - {...register(`watchers.${index}.credentialId`)} + {...register(`localityLbEndpoints.${index}.watcher.kubeconfig.credentialId`)} /> )} + 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. - {fields.map((field, propIndex) => ( - - - - {!readOnly && ( - } - onClick={() => remove(propIndex)} - /> - )} - + ( + + )} + /> + + 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) => ( + mappings.remove(mappingIndex)} + /> ))} {!readOnly && ( )} @@ -431,6 +726,9 @@ interface AggregatorFormFieldsProps { group: string; control: Control; register: UseFormRegister; + setValue: UseFormSetValue; + getValues: UseFormGetValues; + setFocus: UseFormSetFocus; errors: FieldErrors; idReadOnly: boolean; readOnly: boolean; @@ -440,11 +738,14 @@ const AggregatorFormFields = ({ group, control, register, + setValue, + getValues, + setFocus, 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 }); @@ -454,7 +755,7 @@ const AggregatorFormFields = ({ return ( <> - Aggregator ID + remove(index)} canRemove={fields.length > 1} readOnly={readOnly} /> ))} + {!readOnly && ( - + + + )} + + + {/* 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; +}) => { + // 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 ( + + setOpened(!expanded)}> + + Policy (optional) + + {!expanded && ( + + How Envoy balances across these endpoints + + )} + + + + + + Healthy above 100/factor — 140 means 72%. + + + + Weighted priority health + + Weighs priority health by endpoint weight, not count. + + + + + Drops an endpoint unrefreshed for this long. + + + {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 +937,14 @@ const NewK8sAggregatorEditor = ({ group }: { group: string }) => { const { register, control, + setValue, + getValues, + setFocus, handleSubmit, formState: { errors }, - } = useForm({ defaultValues: { aggregatorId: '', watchers: [{ ...emptyWatcher }] } }); + } = useForm({ + defaultValues: { aggregatorId: '', localityLbEndpoints: [{ ...emptyWatcher }], policy: { ...emptyPolicy } }, + }); const onPreview = async (data: FormData) => { setPreviewResult(null); @@ -564,6 +995,9 @@ const NewK8sAggregatorEditor = ({ group }: { group: string }) => { group={group} control={control} register={register} + setValue={setValue} + getValues={getValues} + setFocus={setFocus} errors={errors} idReadOnly={false} readOnly={false} @@ -618,17 +1052,23 @@ const ExistingK8sAggregatorEditor = ({ group, id }: { group: string; id: string const { register, control, + setValue, + getValues, + setFocus, 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 +1083,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 +1180,9 @@ const ExistingK8sAggregatorEditor = ({ group, id }: { group: string; id: string group={group} control={control} register={register} + setValue={setValue} + getValues={getValues} + setFocus={setFocus} 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'); + }); +}); 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); }