From 101339308b18bc12e8feb7330bcbe42b83dddff1 Mon Sep 17 00:00:00 2001 From: Dmitry Mikhaylov Date: Tue, 15 Sep 2026 10:54:41 -0700 Subject: [PATCH 1/5] feat(cf-api): Add Event Ledger publication for NVCF function deployment status transitions. Closes #1290 --- .../cloud-functions/nvcf-core/BUILD.bazel | 1 + .../eventledger/EventLedgerClient.java | 215 ++++++++++++++++++ .../function/FunctionDeploymentService.java | 27 +-- .../FunctionStatusTransitionService.java | 55 +++++ .../function/FunctionsRepositoryTest.java | 66 ++++++ .../eventledger/EventLedgerClientTest.java | 201 ++++++++++++++++ .../src/main/resources/application-ncp.yaml | 7 + .../src/main/resources/application.yaml | 10 + 8 files changed, 566 insertions(+), 16 deletions(-) create mode 100644 src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java create mode 100644 src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/function/FunctionStatusTransitionService.java create mode 100644 src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java diff --git a/src/control-plane-services/cloud-functions/nvcf-core/BUILD.bazel b/src/control-plane-services/cloud-functions/nvcf-core/BUILD.bazel index 2bd46f2e89..5f1316b5e0 100644 --- a/src/control-plane-services/cloud-functions/nvcf-core/BUILD.bazel +++ b/src/control-plane-services/cloud-functions/nvcf-core/BUILD.bazel @@ -96,6 +96,7 @@ nvcf_java_library( "@nv_third_party_deps//:io_grpc_grpc_stub", "@nv_third_party_deps//:net_devh_grpc_common_spring_boot", "@nv_third_party_deps//:io_micrometer_micrometer_core", + "@nv_third_party_deps//:io_micrometer_context_propagation", "@nv_third_party_deps//:io_micrometer_micrometer_observation", "@nv_third_party_deps//:io_micrometer_micrometer_registry_prometheus", "@nv_third_party_deps//:io_micrometer_micrometer_tracing", diff --git a/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java b/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java new file mode 100644 index 0000000000..80c01862e3 --- /dev/null +++ b/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java @@ -0,0 +1,215 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed 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 + * + * http://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. + */ +package com.nvidia.nvcf.service.eventledger; + +import com.nvidia.nvcf.persistence.function.entity.FunctionStatus; +import com.nvidia.nvcf.util.NvcfOAuth2ClientUtils; +import io.micrometer.context.ContextSnapshot; +import io.micrometer.context.ContextSnapshotFactory; +import jakarta.annotation.PreDestroy; +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.cloud.context.config.annotation.RefreshScope; +import org.springframework.http.MediaType; +import org.springframework.stereotype.Service; +import org.springframework.web.reactive.function.client.WebClient; +import tools.jackson.databind.json.JsonMapper; + +@Slf4j +@Service +@RefreshScope +public class EventLedgerClient { + + static final String CLIENT_REGISTRATION_ID = "event-ledger"; + static final String CLOUD_EVENTS_PATH = "/v3/ledger/cloudevents"; + static final String CLOUD_EVENTS_CONTENT_TYPE = "application/cloudevents+json"; + + private static final String MESG_UNKNOWN_FUNCTION_STATUS = + "Event Ledger unknown function status: {}, skip publishing."; + private static final String MESG_QUEUE_FULL = + "Event Ledger queue is full: accountId={}, functionId={}, deploymentId={}, status={}"; + private static final String MESG_FAILED_TO_ENQUEUE_EVENT = + "Failed to enqueue Event Ledger event: accountId={}, functionId={}, deploymentId={}, " + + "status={}"; + private static final String MESG_FAILED_TO_PUBLISH_FUNCTION_STATUS = + "Failed to publish function status to Event Ledger: accountId={}, functionId={}, " + + "functionVersionId={}, deploymentId={}, status={}"; + private static final Map EVENT_NAMES = Map.of( + FunctionStatus.DEPLOYING, "Function.Deploying", + FunctionStatus.ACTIVE, "Function.Ready", + FunctionStatus.DEGRADING, "Function.Degrading", + FunctionStatus.DEGRADED, "Function.Degraded", + FunctionStatus.ERROR, "Function.Error", + FunctionStatus.INACTIVE, "Function.Inactive"); + private static final ContextSnapshotFactory CONTEXT_SNAPSHOT_FACTORY = + ContextSnapshotFactory.builder().build(); + + private final Duration timeout; + private final boolean enabled; + private final WebClient webClient; + private final JsonMapper jsonMapper; + private final ExecutorService executor; + + private record FunctionStatusTransition( + UUID functionId, + UUID functionVersionId, + UUID deploymentId, + FunctionStatus previousStatus, + FunctionStatus currentStatus, + Instant persistedAt) { + } + + @Autowired + public EventLedgerClient( + @Value("${nvcf.event-ledger.enabled:false}") boolean enabled, + @Value("${nvcf.event-ledger.base-url:http://event-ledger.nvcf.svc.cluster.local:8080}") + String baseUrl, + @Value("${nvcf.event-ledger.timeout:2s}") Duration timeout, + @Value("${nvcf.event-ledger.publisher-threads:2}") int publisherThreads, + @Value("${nvcf.event-ledger.queue-capacity:1000}") int queueCapacity, + @Value("${spring.security.oauth2.client.registration.event-ledger.client-id:}") + String clientId, + @Value("${spring.security.oauth2.client.registration.event-ledger.client-secret:}") + String clientSecret, + @Value("${spring.security.oauth2.client.registration.event-ledger.scope:}") String scope, + @Value("${spring.security.oauth2.client.provider.event-ledger.token-uri:}") String tokenUri, + WebClient.Builder webClientBuilder, + JsonMapper jsonMapper) { + this(enabled, timeout, enabled + ? authenticatedWebClient( + baseUrl, clientId, clientSecret, scope, tokenUri, webClientBuilder) + : webClientBuilder.baseUrl(baseUrl).build(), jsonMapper, + new ThreadPoolExecutor(publisherThreads, publisherThreads, 0L, TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(queueCapacity), + Thread.ofPlatform().name("event-ledger-publisher-", 0) + .factory(), + new ThreadPoolExecutor.AbortPolicy())); + } + + EventLedgerClient( + boolean enabled, + Duration timeout, + WebClient webClient, + JsonMapper jsonMapper, + ExecutorService executor) { + this.enabled = enabled; + this.timeout = timeout; + this.webClient = webClient; + this.jsonMapper = jsonMapper; + this.executor = executor; + } + + public void publish( + String ncaId, + UUID functionId, + UUID functionVersionId, + UUID deploymentId, + FunctionStatus previousStatus, + FunctionStatus currentStatus, + Instant persistedAt) { + if (!enabled) { + return; + } + + var transition = new FunctionStatusTransition( + functionId, + functionVersionId, + deploymentId, + previousStatus, + currentStatus, + persistedAt); + var eventName = EVENT_NAMES.get(transition.currentStatus()); + if (eventName == null) { + log.warn(MESG_UNKNOWN_FUNCTION_STATUS, transition.currentStatus()); + return; + } + + try { + // Preserve tracing and logging context when publishing on the executor thread. + ContextSnapshot contextSnapshot = CONTEXT_SNAPSHOT_FACTORY.captureAll(); + executor.execute(contextSnapshot.wrap(() -> send(ncaId, transition, eventName))); + } catch (RejectedExecutionException ex) { + log.warn(MESG_QUEUE_FULL, + ncaId, transition.functionId(), transition.deploymentId(), + transition.currentStatus()); + } catch (RuntimeException ex) { + log.warn(MESG_FAILED_TO_ENQUEUE_EVENT, + ncaId, transition.functionId(), transition.deploymentId(), + transition.currentStatus(), ex); + } + } + + private void send( + String ncaId, FunctionStatusTransition transition, String eventName) { + try { + var payload = Map.of( + "specversion", "1.0", + "id", UUID.randomUUID().toString(), + "source", "cloud-functions", + "type", eventName, + "time", transition.persistedAt().toString(), + "namespace", ncaId, + "deploymentId", transition.deploymentId().toString(), + "data", Map.of( + "functionId", transition.functionId().toString(), + "functionVersionId", transition.functionVersionId().toString(), + "deploymentId", transition.deploymentId().toString(), + "previousStatus", transition.previousStatus().toString(), + "currentStatus", transition.currentStatus().toString())); + + webClient.post() + .uri(CLOUD_EVENTS_PATH) + .contentType(MediaType.parseMediaType(CLOUD_EVENTS_CONTENT_TYPE)) + .bodyValue(jsonMapper.writeValueAsBytes(payload)) + .retrieve() + .toBodilessEntity() + .block(timeout); + } catch (Exception ex) { + log.warn(MESG_FAILED_TO_PUBLISH_FUNCTION_STATUS, + ncaId, transition.functionId(), transition.functionVersionId(), + transition.deploymentId(), transition.currentStatus(), ex); + } + } + + private static WebClient authenticatedWebClient( + String baseUrl, + String clientId, + String clientSecret, + String scope, + String tokenUri, + WebClient.Builder builder) { + return builder.baseUrl(baseUrl) + .filter(NvcfOAuth2ClientUtils.getOAuth2ExchangeFilter( + builder, CLIENT_REGISTRATION_ID, tokenUri, clientId, clientSecret, scope)) + .build(); + } + + @PreDestroy + void close() { + executor.shutdownNow(); + } +} diff --git a/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/function/FunctionDeploymentService.java b/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/function/FunctionDeploymentService.java index 80ce223295..21fb41b29a 100644 --- a/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/function/FunctionDeploymentService.java +++ b/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/function/FunctionDeploymentService.java @@ -145,6 +145,7 @@ public class FunctionDeploymentService { private final JsonMapper jsonMapper; private final WorkerNatsService workerNatsService; private final RegistryArtifactService artifactService; + private final FunctionStatusTransitionService functionStatusTransitionService; // Function deployment is created by the admin. @SneakyThrows @@ -228,13 +229,12 @@ public FunctionDeploymentDto createFunctionDeployment( // functions_deployment_v2 and gpu_specifications in one batch. deploymentBatchWriter.createDeployment( new FunctionDeploymentContext(deployment, gpuSpecEntities)); + var newStatus = DEPLOYING; if (isZeroScaling) { log.info(MESG_ZERO_SCALE, functionId, functionVersionId); - function.setFunctionStatus(ACTIVE); - } else { - function.setFunctionStatus(DEPLOYING); + newStatus = ACTIVE; } - functionsRepository.insert(function); + functionStatusTransitionService.persist(function, deploymentId, newStatus); } catch (Exception ex) { var mesg = format(MESG_FAILED_CREATE_DEPLOYMENT, functionId, functionVersionId, status, ex.getMessage()); @@ -321,8 +321,8 @@ private FunctionDto deleteFunctionDeployment( try { // Mark function as INACTIVE to stop accepting new jobs/tasks for this version. - function.setFunctionStatus(INACTIVE); - functionsRepository.insert(function); + functionStatusTransitionService.persist( + function, deployment.getDeploymentId(), INACTIVE); // The current default behavior for deleting a deployment is ungraceful as we whack the // queue and the workers. In the future, when we get the opportunity to break backward @@ -482,8 +482,7 @@ public FunctionEntity transitionFunctionToActive( }); var jsonBefore = jsonMapper.valueToTree(function); - function.setFunctionStatus(ACTIVE); - functionsRepository.insert(function); + functionStatusTransitionService.persist(function, deploymentId, ACTIVE); var summary = SUMMARY_ACTIVATE_FUNCTION.formatted(functionId, functionVersionId); functionAuditService.auditFunctionUpdate(summary, STATE_ACTIVATED, jsonBefore, function); @@ -502,8 +501,7 @@ public FunctionEntity transitionDeployingFunctionToError( // Once we set the function status to ERROR we'll no longer attempt to // clean its deployment. Org Admin should delete the deployment to reset // the function's status to ACTIVE. - function.setFunctionStatus(FunctionStatus.ERROR); - functionsRepository.insert(function); + functionStatusTransitionService.persist(function, deploymentId, FunctionStatus.ERROR); var summary = SUMMARY_ERROR_FUNCTION.formatted(functionId, functionVersionId); functionAuditService.auditFunctionUpdate(summary, STATE_ERROR, jsonBefore, function); @@ -529,8 +527,7 @@ public FunctionEntity transitionFunctionToDegrading( }); var jsonBefore = jsonMapper.valueToTree(function); - function.setFunctionStatus(DEGRADING); - functionsRepository.insert(function); + functionStatusTransitionService.persist(function, deploymentId, DEGRADING); var summary = SUMMARY_DEGRADING_FUNCTION.formatted(functionId, functionVersionId); functionAuditService.auditFunctionUpdate(summary, STATE_DEGRADING, jsonBefore, function); @@ -556,8 +553,7 @@ public FunctionEntity transitionFunctionToDegraded( }); var jsonBefore = jsonMapper.valueToTree(function); - function.setFunctionStatus(DEGRADED); - functionsRepository.insert(function); + functionStatusTransitionService.persist(function, deploymentId, DEGRADED); var summary = SUMMARY_DEGRADED_FUNCTION.formatted(functionId, functionVersionId); functionAuditService.auditFunctionUpdate(summary, STATE_DEGRADED, jsonBefore, function); @@ -582,8 +578,7 @@ private FunctionEntity transitionFunctionToInactive( }); var jsonBefore = jsonMapper.valueToTree(function); - function.setFunctionStatus(INACTIVE); - functionsRepository.insert(function); + functionStatusTransitionService.persist(function, deploymentId, INACTIVE); var summary = SUMMARY_INACTIVATE_FUNCTION.formatted(functionId, functionVersionId); functionAuditService.auditFunctionUpdate(summary, STATE_INACTIVE, jsonBefore, function); diff --git a/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/function/FunctionStatusTransitionService.java b/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/function/FunctionStatusTransitionService.java new file mode 100644 index 0000000000..54f1cce4c2 --- /dev/null +++ b/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/function/FunctionStatusTransitionService.java @@ -0,0 +1,55 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed 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 + * + * http://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. + */ +package com.nvidia.nvcf.service.function; + +import com.nvidia.nvcf.persistence.function.FunctionsRepository; +import com.nvidia.nvcf.persistence.function.entity.FunctionEntity; +import com.nvidia.nvcf.persistence.function.entity.FunctionStatus; +import com.nvidia.nvcf.service.eventledger.EventLedgerClient; +import java.time.Clock; +import java.time.Instant; +import java.util.UUID; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; + +@Service +@RequiredArgsConstructor +public class FunctionStatusTransitionService { + + private final FunctionsRepository functionsRepository; + private final EventLedgerClient eventLedgerClient; + // Inject the time source so transition timestamps can be controlled in tests. + private final Clock clock; + + public void persist( + FunctionEntity function, UUID deploymentId, FunctionStatus newStatus) { + var previousStatus = function.getFunctionStatus(); + if (previousStatus == newStatus) { + return; + } + function.setFunctionStatus(newStatus); + functionsRepository.insert(function); + eventLedgerClient.publish( + function.getNcaId(), + function.getFunctionId(), + function.getFunctionVersionId(), + deploymentId, + previousStatus, + newStatus, + Instant.now(clock)); + } +} diff --git a/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/persistence/function/FunctionsRepositoryTest.java b/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/persistence/function/FunctionsRepositoryTest.java index bbad4f9db0..d90ac81b61 100644 --- a/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/persistence/function/FunctionsRepositoryTest.java +++ b/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/persistence/function/FunctionsRepositoryTest.java @@ -51,6 +51,11 @@ import static java.util.concurrent.Future.State.RUNNING; import static java.util.concurrent.Future.State.SUCCESS; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import com.google.common.collect.Sets; import com.nvidia.nvcf.IntegrationTestConfiguration; @@ -62,8 +67,11 @@ import com.nvidia.nvcf.persistence.function.entity.GpuSpecificationEntity; import com.nvidia.nvcf.persistence.function.entity.GpuSpecificationKey; import com.nvidia.nvcf.rest.function.deployment.dto.HelmValidationPolicyDto; +import com.nvidia.nvcf.service.eventledger.EventLedgerClient; import com.nvidia.nvcf.service.function.FunctionDeploymentContext; import com.nvidia.nvcf.service.function.FunctionDeploymentLookupService; +import com.nvidia.nvcf.service.function.FunctionStatusTransitionService; +import java.time.Clock; import java.time.Instant; import java.util.List; import java.util.Map; @@ -80,9 +88,11 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInstance; +import org.mockito.ArgumentCaptor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.bean.override.mockito.MockitoBean; import tools.jackson.databind.json.JsonMapper; @TestInstance(TestInstance.Lifecycle.PER_CLASS) @@ -111,6 +121,15 @@ class FunctionsRepositoryTest { @Autowired private JsonMapper jsonMapper; + @Autowired + private FunctionStatusTransitionService functionStatusTransitionService; + + @Autowired + private Clock clock; + + @MockitoBean + private EventLedgerClient eventLedgerClient; + private Set functionLevelAuthzParties = Set.of(TEST_AUTHORIZED_NCA_ID_1, TEST_AUTHORIZED_NCA_ID_2); private Set versionLevelAuthzParties = Set.of(TEST_AUTHORIZED_NCA_ID_4, @@ -130,6 +149,7 @@ void cleanup() { void init() { functionsRepository.deleteAll(); functionsDeploymentRepository.deleteAll(); + clearInvocations(eventLedgerClient); } @AfterEach @@ -535,6 +555,52 @@ void testListIndexFetchByFunctionId() { .containsExactlyInAnyOrder(TEST_VERSION_ID_3); } + @Test + void persistsFunctionStatusTransitionAndPublishesEvent() { + var function = getTestEntity( + TEST_FUNCTION_ID, TEST_VERSION_ID_1, TEST_NCA_ID, TEST_FUNCTION_NAME); + functionsRepository.save(function); + var timestampLowerBound = Instant.now(clock); + + functionStatusTransitionService.persist( + function, TEST_DEPLOYMENT_ID, FunctionStatus.ACTIVE); + var timestampUpperBound = Instant.now(clock); + + var persistedFunction = functionsRepository + .getByFunctionVersionId(TEST_VERSION_ID_1) + .orElseThrow(); + assertThat(persistedFunction.getFunctionStatus()).isEqualTo(FunctionStatus.ACTIVE); + + var timestampCaptor = ArgumentCaptor.forClass(Instant.class); + verify(eventLedgerClient).publish( + eq(TEST_NCA_ID), + eq(TEST_FUNCTION_ID), + eq(TEST_VERSION_ID_1), + eq(TEST_DEPLOYMENT_ID), + eq(FunctionStatus.INACTIVE), + eq(FunctionStatus.ACTIVE), + timestampCaptor.capture()); + assertThat(timestampCaptor.getValue()) + .isBetween(timestampLowerBound, timestampUpperBound); + } + + @Test + void skipsUnchangedFunctionStatus() { + var function = getTestEntity( + TEST_FUNCTION_ID, TEST_VERSION_ID_1, TEST_NCA_ID, TEST_FUNCTION_NAME); + functionsRepository.save(function); + + functionStatusTransitionService.persist( + function, TEST_DEPLOYMENT_ID, FunctionStatus.INACTIVE); + + assertThat(functionsRepository.getByFunctionVersionId(TEST_VERSION_ID_1)) + .get() + .extracting(FunctionEntity::getFunctionStatus) + .isEqualTo(FunctionStatus.INACTIVE); + verify(eventLedgerClient, never()).publish( + eq(TEST_NCA_ID), any(), any(), any(), any(), any(), any()); + } + private FunctionEntity getTestEntity( UUID id, UUID versionId, diff --git a/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java b/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java new file mode 100644 index 0000000000..6f378d34e9 --- /dev/null +++ b/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java @@ -0,0 +1,201 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed 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 + * + * http://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. + */ +package com.nvidia.nvcf.service.eventledger; + +import static com.github.tomakehurst.wiremock.client.WireMock.aResponse; +import static com.github.tomakehurst.wiremock.client.WireMock.equalTo; +import static com.github.tomakehurst.wiremock.client.WireMock.post; +import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo; +import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.options; +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.params.provider.Arguments.arguments; + +import com.github.tomakehurst.wiremock.WireMockServer; +import com.nvidia.nvcf.persistence.function.entity.FunctionStatus; +import java.time.Duration; +import java.time.Instant; +import java.util.UUID; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.convert.ApplicationConversionService; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.cloud.autoconfigure.RefreshAutoConfiguration; +import org.springframework.web.reactive.function.client.WebClient; +import tools.jackson.databind.json.JsonMapper; + +class EventLedgerClientTest { + + private WireMockServer server; + private EventLedgerClient client; + + @BeforeEach + void setUp() { + server = new WireMockServer(options().dynamicPort()); + server.start(); + client = newClient(Duration.ofSeconds(1)); + } + + @AfterEach + void tearDown() { + client.close(); + server.stop(); + } + + @ParameterizedTest + @MethodSource("statusMappings") + void publishesStructuredCloudEvent(FunctionStatus status, String eventName) throws Exception { + server.stubFor(post(urlEqualTo(EventLedgerClient.CLOUD_EVENTS_PATH)) + .willReturn(aResponse().withStatus(202))); + var previousStatus = status == FunctionStatus.INACTIVE + ? FunctionStatus.ERROR + : FunctionStatus.INACTIVE; + var functionId = UUID.randomUUID(); + var functionVersionId = UUID.randomUUID(); + var deploymentId = UUID.randomUUID(); + var persistedAt = Instant.parse("2026-09-08T12:34:56Z"); + + client.publish( + "account-1", functionId, functionVersionId, deploymentId, + previousStatus, status, persistedAt); + + await().untilAsserted(() -> assertThat(server.getAllServeEvents()).hasSize(1)); + var request = server.getAllServeEvents().getFirst().getRequest(); + var payload = new JsonMapper().readTree(request.getBody()); + assertThat(request.getHeader("Content-Type")) + .isEqualTo(EventLedgerClient.CLOUD_EVENTS_CONTENT_TYPE); + assertThat(payload.get("specversion").asText()).isEqualTo("1.0"); + assertThat(payload.get("id").asText()).isNotBlank(); + assertThat(payload.get("source").asText()).isEqualTo("cloud-functions"); + assertThat(payload.get("type").asText()).isEqualTo(eventName); + assertThat(payload.get("time").asText()).isEqualTo(persistedAt.toString()); + assertThat(payload.get("namespace").asText()).isEqualTo("account-1"); + assertThat(payload.get("deploymentId").asText()) + .isEqualTo(deploymentId.toString()); + assertThat(payload.get("data").get("functionId").asText()) + .isEqualTo(functionId.toString()); + assertThat(payload.get("data").get("functionVersionId").asText()) + .isEqualTo(functionVersionId.toString()); + assertThat(payload.get("data").get("deploymentId").asText()) + .isEqualTo(deploymentId.toString()); + assertThat(payload.get("data").get("previousStatus").asText()) + .isEqualTo(previousStatus.toString()); + assertThat(payload.get("data").get("currentStatus").asText()) + .isEqualTo(status.toString()); + } + + @Test + void springCreatesEnabledClientUsingProductionConstructor() { + new ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of(RefreshAutoConfiguration.class)) + .withInitializer(context -> context.getBeanFactory() + .setConversionService(ApplicationConversionService.getSharedInstance())) + .withPropertyValues( + "nvcf.event-ledger.enabled=true", + "spring.security.oauth2.client.registration.event-ledger.client-id=test", + "spring.security.oauth2.client.registration.event-ledger.client-secret=test", + "spring.security.oauth2.client.registration.event-ledger.scope=test", + "spring.security.oauth2.client.provider.event-ledger.token-uri=http://token") + .withBean(WebClient.Builder.class, WebClient::builder) + .withBean(JsonMapper.class, JsonMapper::new) + .withUserConfiguration(EventLedgerClient.class) + .run(context -> { + assertThat(context).hasNotFailed(); + assertThat(context.getBean("scopedTarget.eventLedgerClient")) + .isInstanceOf(EventLedgerClient.class); + }); + } + + @Test + void springCreatesDisabledClientWithoutOAuthConfiguration() { + new ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of(RefreshAutoConfiguration.class)) + .withInitializer(context -> context.getBeanFactory() + .setConversionService(ApplicationConversionService.getSharedInstance())) + .withBean(WebClient.Builder.class, WebClient::builder) + .withBean(JsonMapper.class, JsonMapper::new) + .withUserConfiguration(EventLedgerClient.class) + .run(context -> { + assertThat(context).hasNotFailed(); + assertThat(context.getBean("scopedTarget.eventLedgerClient")) + .isInstanceOf(EventLedgerClient.class); + }); + } + + @Test + void handlesPublishFailureWithoutThrowing() { + server.stubFor(post(urlEqualTo(EventLedgerClient.CLOUD_EVENTS_PATH)) + .withHeader("Content-Type", + equalTo(EventLedgerClient.CLOUD_EVENTS_CONTENT_TYPE)) + .willReturn(aResponse().withStatus(500))); + + publish(FunctionStatus.DEPLOYING, FunctionStatus.ERROR); + + await().untilAsserted(() -> assertThat(server.getAllServeEvents()).hasSize(1)); + } + + @Test + void skipsPublishingWhenDisabled() { + client.close(); + client = newClient(Duration.ofSeconds(1), false); + + publish(FunctionStatus.DEPLOYING, FunctionStatus.ACTIVE); + + assertThat(server.getAllServeEvents()).isEmpty(); + } + + private EventLedgerClient newClient(Duration timeout) { + return newClient(timeout, true); + } + + private EventLedgerClient newClient(Duration timeout, boolean enabled) { + var executor = new ThreadPoolExecutor( + 1, 1, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(10)); + return new EventLedgerClient( + enabled, + timeout, + WebClient.builder().baseUrl(server.baseUrl()).build(), + new JsonMapper(), + executor); + } + + private void publish( + FunctionStatus previousStatus, FunctionStatus currentStatus) { + client.publish( + "account-1", UUID.randomUUID(), UUID.randomUUID(), UUID.randomUUID(), + previousStatus, currentStatus, Instant.parse("2026-09-08T12:34:56Z")); + } + + private static Stream statusMappings() { + return Stream.of( + arguments(FunctionStatus.DEPLOYING, "Function.Deploying"), + arguments(FunctionStatus.ACTIVE, "Function.Ready"), + arguments(FunctionStatus.DEGRADING, "Function.Degrading"), + arguments(FunctionStatus.DEGRADED, "Function.Degraded"), + arguments(FunctionStatus.ERROR, "Function.Error"), + arguments(FunctionStatus.INACTIVE, "Function.Inactive")); + } +} diff --git a/src/control-plane-services/cloud-functions/nvcf-service/src/main/resources/application-ncp.yaml b/src/control-plane-services/cloud-functions/nvcf-service/src/main/resources/application-ncp.yaml index 4f212ecb48..1e61577046 100644 --- a/src/control-plane-services/cloud-functions/nvcf-service/src/main/resources/application-ncp.yaml +++ b/src/control-plane-services/cloud-functions/nvcf-service/src/main/resources/application-ncp.yaml @@ -42,6 +42,9 @@ spring: reval: client-id: unused client-secret: unused + event-ledger: + client-id: unused + client-secret: unused provider: api-keys: token-uri: unused @@ -53,6 +56,8 @@ spring: token-uri: unused reval: token-uri: unused + event-ledger: + token-uri: unused management: tracing: @@ -127,6 +132,8 @@ nvcf: # base-url: http://api.icms.svc.cluster.local:8080 -- Use when self-hosted stack is ready static: token: ${kv.tokens.icms} + event-ledger: + enabled: false api-keys: base-url: http://api-keys.api-keys.svc.cluster.local:8080 # base-url: http://api-keys.nvcf.svc.cluster.local:8080 -- Use when self-hosted stack is ready diff --git a/src/control-plane-services/cloud-functions/nvcf-service/src/main/resources/application.yaml b/src/control-plane-services/cloud-functions/nvcf-service/src/main/resources/application.yaml index e18fc0d68e..0d99eaa1fc 100644 --- a/src/control-plane-services/cloud-functions/nvcf-service/src/main/resources/application.yaml +++ b/src/control-plane-services/cloud-functions/nvcf-service/src/main/resources/application.yaml @@ -94,6 +94,12 @@ spring: client-secret: ${kv.oauth2.client-secret} authorization-grant-type: client_credentials scope: helmreval:validate + event-ledger: + provider: event-ledger + client-id: ${kv.oauth2.client-id} + client-secret: ${kv.oauth2.client-secret} + authorization-grant-type: client_credentials + scope: fnds:createEvent provider: icms: token-uri: ${spring.security.oauth2.resourceserver.jwt.issuer-uri}/token @@ -105,6 +111,8 @@ spring: token-uri: ${spring.security.oauth2.resourceserver.jwt.issuer-uri}/token reval: token-uri: ${spring.security.oauth2.resourceserver.jwt.issuer-uri}/token + event-ledger: + token-uri: ${spring.security.oauth2.resourceserver.jwt.issuer-uri}/token logging: level: @@ -310,6 +318,8 @@ nvcf: icms: allocator: maximum-target-latency: PT10S + event-ledger: + enabled: true notary: jwt: issuer-uri: ${nvcf.notary.base-url} From 558e136d46133251a4a67f33c251ee8fb5101af4 Mon Sep 17 00:00:00 2001 From: Dmitry Mikhaylov Date: Tue, 15 Sep 2026 11:46:46 -0700 Subject: [PATCH 2/5] rebuild event --- .../cloud-functions/nvcf-core/BUILD.bazel | 5 ++ .../eventledger/EventLedgerClient.java | 52 +++++++++++++------ .../eventledger/EventLedgerClientTest.java | 38 ++++++++------ 3 files changed, 62 insertions(+), 33 deletions(-) diff --git a/src/control-plane-services/cloud-functions/nvcf-core/BUILD.bazel b/src/control-plane-services/cloud-functions/nvcf-core/BUILD.bazel index 5f1316b5e0..9385b279e9 100644 --- a/src/control-plane-services/cloud-functions/nvcf-core/BUILD.bazel +++ b/src/control-plane-services/cloud-functions/nvcf-core/BUILD.bazel @@ -94,6 +94,9 @@ nvcf_java_library( "@nv_third_party_deps//:io_grpc_grpc_api", "@nv_third_party_deps//:io_grpc_grpc_netty_shaded", "@nv_third_party_deps//:io_grpc_grpc_stub", + "@nv_third_party_deps//:io_cloudevents_cloudevents_api", + "@nv_third_party_deps//:io_cloudevents_cloudevents_core", + "@nv_third_party_deps//:io_cloudevents_cloudevents_json_jackson", "@nv_third_party_deps//:net_devh_grpc_common_spring_boot", "@nv_third_party_deps//:io_micrometer_micrometer_core", "@nv_third_party_deps//:io_micrometer_context_propagation", @@ -191,6 +194,8 @@ NVCF_CORE_TEST_DEPS = [ "@nv_third_party_deps//:io_grpc_grpc_api", "@nv_third_party_deps//:io_grpc_grpc_services", "@nv_third_party_deps//:io_grpc_grpc_stub", + "@nv_third_party_deps//:io_cloudevents_cloudevents_core", + "@nv_third_party_deps//:io_cloudevents_cloudevents_json_jackson", "@nv_third_party_deps//:io_micrometer_micrometer_core", "@nv_third_party_deps//:io_micrometer_micrometer_observation", "@nv_third_party_deps//:io_micrometer_micrometer_tracing", diff --git a/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java b/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java index 80c01862e3..0b22cce8c3 100644 --- a/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java +++ b/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java @@ -16,13 +16,22 @@ */ package com.nvidia.nvcf.service.eventledger; +import static io.cloudevents.jackson.JsonFormat.CONTENT_TYPE; + import com.nvidia.nvcf.persistence.function.entity.FunctionStatus; import com.nvidia.nvcf.util.NvcfOAuth2ClientUtils; +import io.cloudevents.CloudEvent; +import io.cloudevents.core.builder.CloudEventBuilder; +import io.cloudevents.core.format.EventFormat; +import io.cloudevents.core.provider.EventFormatProvider; import io.micrometer.context.ContextSnapshot; import io.micrometer.context.ContextSnapshotFactory; import jakarta.annotation.PreDestroy; +import java.net.URI; +import java.nio.charset.StandardCharsets; import java.time.Duration; import java.time.Instant; +import java.time.ZoneOffset; import java.util.Map; import java.util.UUID; import java.util.concurrent.ArrayBlockingQueue; @@ -46,7 +55,9 @@ public class EventLedgerClient { static final String CLIENT_REGISTRATION_ID = "event-ledger"; static final String CLOUD_EVENTS_PATH = "/v3/ledger/cloudevents"; - static final String CLOUD_EVENTS_CONTENT_TYPE = "application/cloudevents+json"; + static final String CLOUD_EVENTS_CONTENT_TYPE = CONTENT_TYPE; + static final String CLOUD_EVENT_SOURCE = "nvidia-spot"; + static final String CLOUD_EVENT_TYPE = "nvcf-api"; private static final String MESG_UNKNOWN_FUNCTION_STATUS = "Event Ledger unknown function status: {}, skip publishing."; @@ -73,6 +84,7 @@ public class EventLedgerClient { private final WebClient webClient; private final JsonMapper jsonMapper; private final ExecutorService executor; + private final EventFormat eventFormat; private record FunctionStatusTransition( UUID functionId, @@ -121,6 +133,7 @@ public EventLedgerClient( this.webClient = webClient; this.jsonMapper = jsonMapper; this.executor = executor; + this.eventFormat = EventFormatProvider.getInstance().resolveFormat(CONTENT_TYPE); } public void publish( @@ -166,25 +179,10 @@ public void publish( private void send( String ncaId, FunctionStatusTransition transition, String eventName) { try { - var payload = Map.of( - "specversion", "1.0", - "id", UUID.randomUUID().toString(), - "source", "cloud-functions", - "type", eventName, - "time", transition.persistedAt().toString(), - "namespace", ncaId, - "deploymentId", transition.deploymentId().toString(), - "data", Map.of( - "functionId", transition.functionId().toString(), - "functionVersionId", transition.functionVersionId().toString(), - "deploymentId", transition.deploymentId().toString(), - "previousStatus", transition.previousStatus().toString(), - "currentStatus", transition.currentStatus().toString())); - webClient.post() .uri(CLOUD_EVENTS_PATH) .contentType(MediaType.parseMediaType(CLOUD_EVENTS_CONTENT_TYPE)) - .bodyValue(jsonMapper.writeValueAsBytes(payload)) + .bodyValue(eventFormat.serialize(buildCloudEvent(ncaId, transition, eventName))) .retrieve() .toBodilessEntity() .block(timeout); @@ -195,6 +193,26 @@ private void send( } } + private CloudEvent buildCloudEvent( + String ncaId, FunctionStatusTransition transition, String eventName) throws Exception { + var details = Map.of( + "previousStatus", transition.previousStatus().toString(), + "currentStatus", transition.currentStatus().toString()); + return CloudEventBuilder.v1() + .withId(UUID.randomUUID().toString()) + .withSource(URI.create(CLOUD_EVENT_SOURCE)) + .withTime(transition.persistedAt().atOffset(ZoneOffset.UTC)) + .withType(eventName) + .withExtension("namespace", transition.functionVersionId().toString()) + .withExtension("functionid", transition.functionId().toString()) + .withExtension("functionversionid", transition.functionVersionId().toString()) + .withExtension("deploymentid", transition.deploymentId().toString()) + .withExtension("ncaid", ncaId) + .withExtension("eventtype", CLOUD_EVENT_TYPE) + .withData(jsonMapper.writeValueAsString(details).getBytes(StandardCharsets.UTF_8)) + .build(); + } + private static WebClient authenticatedWebClient( String baseUrl, String clientId, diff --git a/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java b/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java index 6f378d34e9..958f6e65b3 100644 --- a/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java +++ b/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java @@ -21,14 +21,17 @@ import static com.github.tomakehurst.wiremock.client.WireMock.post; import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo; import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.options; +import static io.cloudevents.jackson.JsonFormat.CONTENT_TYPE; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; import static org.junit.jupiter.params.provider.Arguments.arguments; import com.github.tomakehurst.wiremock.WireMockServer; import com.nvidia.nvcf.persistence.function.entity.FunctionStatus; +import io.cloudevents.core.provider.EventFormatProvider; import java.time.Duration; import java.time.Instant; +import java.time.ZoneOffset; import java.util.UUID; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; @@ -84,26 +87,29 @@ void publishesStructuredCloudEvent(FunctionStatus status, String eventName) thro await().untilAsserted(() -> assertThat(server.getAllServeEvents()).hasSize(1)); var request = server.getAllServeEvents().getFirst().getRequest(); - var payload = new JsonMapper().readTree(request.getBody()); + var cloudEvent = EventFormatProvider.getInstance() + .resolveFormat(CONTENT_TYPE) + .deserialize(request.getBody()); assertThat(request.getHeader("Content-Type")) .isEqualTo(EventLedgerClient.CLOUD_EVENTS_CONTENT_TYPE); - assertThat(payload.get("specversion").asText()).isEqualTo("1.0"); - assertThat(payload.get("id").asText()).isNotBlank(); - assertThat(payload.get("source").asText()).isEqualTo("cloud-functions"); - assertThat(payload.get("type").asText()).isEqualTo(eventName); - assertThat(payload.get("time").asText()).isEqualTo(persistedAt.toString()); - assertThat(payload.get("namespace").asText()).isEqualTo("account-1"); - assertThat(payload.get("deploymentId").asText()) - .isEqualTo(deploymentId.toString()); - assertThat(payload.get("data").get("functionId").asText()) - .isEqualTo(functionId.toString()); - assertThat(payload.get("data").get("functionVersionId").asText()) + assertThat(cloudEvent.getSpecVersion().toString()).isEqualTo("1.0"); + assertThat(cloudEvent.getId()).isNotBlank(); + assertThat(cloudEvent.getSource().toString()) + .isEqualTo(EventLedgerClient.CLOUD_EVENT_SOURCE); + assertThat(cloudEvent.getType()).isEqualTo(eventName); + assertThat(cloudEvent.getTime()).isEqualTo(persistedAt.atOffset(ZoneOffset.UTC)); + assertThat(cloudEvent.getExtension("namespace")).isEqualTo(functionVersionId.toString()); + assertThat(cloudEvent.getExtension("functionid")).isEqualTo(functionId.toString()); + assertThat(cloudEvent.getExtension("functionversionid")) .isEqualTo(functionVersionId.toString()); - assertThat(payload.get("data").get("deploymentId").asText()) - .isEqualTo(deploymentId.toString()); - assertThat(payload.get("data").get("previousStatus").asText()) + assertThat(cloudEvent.getExtension("deploymentid")).isEqualTo(deploymentId.toString()); + assertThat(cloudEvent.getExtension("ncaid")).isEqualTo("account-1"); + assertThat(cloudEvent.getExtension("eventtype")) + .isEqualTo(EventLedgerClient.CLOUD_EVENT_TYPE); + var data = new JsonMapper().readTree(cloudEvent.getData().toBytes()); + assertThat(data.get("previousStatus").asText()) .isEqualTo(previousStatus.toString()); - assertThat(payload.get("data").get("currentStatus").asText()) + assertThat(data.get("currentStatus").asText()) .isEqualTo(status.toString()); } From 4222d03ea7b42dc7cac3c47eb90728c5b23aa6ac Mon Sep 17 00:00:00 2001 From: Dmitry Mikhaylov Date: Tue, 15 Sep 2026 12:28:57 -0700 Subject: [PATCH 3/5] updated NOTICE file --- src/control-plane-services/cloud-functions/NOTICE | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/control-plane-services/cloud-functions/NOTICE b/src/control-plane-services/cloud-functions/NOTICE index 03defd4f5e..9883b49bdf 100644 --- a/src/control-plane-services/cloud-functions/NOTICE +++ b/src/control-plane-services/cloud-functions/NOTICE @@ -1,5 +1,5 @@ -Lists of 288 third-party dependencies. +Lists of 291 third-party dependencies. (Apache License, Version 2.0) LZ4 Java Compression (at.yawk.lz4:lz4-java:1.11.2 - https://github.com/yawkat/lz4-java) (EPL-2.0) (LGPL-2.1-only) Logback Classic Module (ch.qos.logback:logback-classic:1.5.38 - http://logback.qos.ch) (EPL-2.0) (LGPL-2.1-only) Logback Core Module (ch.qos.logback:logback-core:1.5.38 - http://logback.qos.ch) @@ -48,6 +48,9 @@ Lists of 288 third-party dependencies. (Apache-2.0) Apache Commons Codec (commons-codec:commons-codec:1.21.0 - https://commons.apache.org/proper/commons-codec/) (Apache-2.0) Apache Commons IO (commons-io:commons-io:2.20.0 - https://commons.apache.org/proper/commons-io/) (Apache-2.0) Apache Commons Logging (commons-logging:commons-logging:1.4.0 - https://commons.apache.org/proper/commons-logging/) + (The Apache Software License, Version 2.0) CloudEvents - API (io.cloudevents:cloudevents-api:4.1.1 - https://cloudevents.github.io/sdk-java/) + (The Apache Software License, Version 2.0) CloudEvents - Core (io.cloudevents:cloudevents-core:4.1.1 - https://cloudevents.github.io/sdk-java/) + (The Apache Software License, Version 2.0) CloudEvents - JSON Jackson (io.cloudevents:cloudevents-json-jackson:4.1.1 - https://cloudevents.github.io/sdk-java/) (Apache 2.0) io.grpc:grpc-api (io.grpc:grpc-api:1.83.1 - https://github.com/grpc/grpc-java) (Apache 2.0) io.grpc:grpc-context (io.grpc:grpc-context:1.83.1 - https://github.com/grpc/grpc-java) (Apache 2.0) io.grpc:grpc-core (io.grpc:grpc-core:1.83.1 - https://github.com/grpc/grpc-java) From 883a46ef6792ce39154d0742b0034def5409adbe Mon Sep 17 00:00:00 2001 From: Dmitry Mikhaylov Date: Tue, 15 Sep 2026 16:14:55 -0700 Subject: [PATCH 4/5] fixed source field --- .../nvidia/nvcf/service/eventledger/EventLedgerClient.java | 4 +--- .../nvcf/service/eventledger/EventLedgerClientTest.java | 2 -- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java b/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java index 0b22cce8c3..041d2f72be 100644 --- a/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java +++ b/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java @@ -56,8 +56,7 @@ public class EventLedgerClient { static final String CLIENT_REGISTRATION_ID = "event-ledger"; static final String CLOUD_EVENTS_PATH = "/v3/ledger/cloudevents"; static final String CLOUD_EVENTS_CONTENT_TYPE = CONTENT_TYPE; - static final String CLOUD_EVENT_SOURCE = "nvidia-spot"; - static final String CLOUD_EVENT_TYPE = "nvcf-api"; + static final String CLOUD_EVENT_SOURCE = "nvidia-cloud-functions"; private static final String MESG_UNKNOWN_FUNCTION_STATUS = "Event Ledger unknown function status: {}, skip publishing."; @@ -208,7 +207,6 @@ private CloudEvent buildCloudEvent( .withExtension("functionversionid", transition.functionVersionId().toString()) .withExtension("deploymentid", transition.deploymentId().toString()) .withExtension("ncaid", ncaId) - .withExtension("eventtype", CLOUD_EVENT_TYPE) .withData(jsonMapper.writeValueAsString(details).getBytes(StandardCharsets.UTF_8)) .build(); } diff --git a/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java b/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java index 958f6e65b3..b675040590 100644 --- a/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java +++ b/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java @@ -104,8 +104,6 @@ void publishesStructuredCloudEvent(FunctionStatus status, String eventName) thro .isEqualTo(functionVersionId.toString()); assertThat(cloudEvent.getExtension("deploymentid")).isEqualTo(deploymentId.toString()); assertThat(cloudEvent.getExtension("ncaid")).isEqualTo("account-1"); - assertThat(cloudEvent.getExtension("eventtype")) - .isEqualTo(EventLedgerClient.CLOUD_EVENT_TYPE); var data = new JsonMapper().readTree(cloudEvent.getData().toBytes()); assertThat(data.get("previousStatus").asText()) .isEqualTo(previousStatus.toString()); From 2fcc476c4ac0ee07ba93343caa6cad6b138b58ea Mon Sep 17 00:00:00 2001 From: Dmitry Mikhaylov Date: Fri, 18 Sep 2026 11:59:30 -0700 Subject: [PATCH 5/5] make Event Ledger call synchronous --- .../cloud-functions/nvcf-core/BUILD.bazel | 1 - .../eventledger/EventLedgerClient.java | 49 ++----------------- .../eventledger/EventLedgerClientTest.java | 28 ++++++----- 3 files changed, 19 insertions(+), 59 deletions(-) diff --git a/src/control-plane-services/cloud-functions/nvcf-core/BUILD.bazel b/src/control-plane-services/cloud-functions/nvcf-core/BUILD.bazel index 9385b279e9..a35ab860cf 100644 --- a/src/control-plane-services/cloud-functions/nvcf-core/BUILD.bazel +++ b/src/control-plane-services/cloud-functions/nvcf-core/BUILD.bazel @@ -99,7 +99,6 @@ nvcf_java_library( "@nv_third_party_deps//:io_cloudevents_cloudevents_json_jackson", "@nv_third_party_deps//:net_devh_grpc_common_spring_boot", "@nv_third_party_deps//:io_micrometer_micrometer_core", - "@nv_third_party_deps//:io_micrometer_context_propagation", "@nv_third_party_deps//:io_micrometer_micrometer_observation", "@nv_third_party_deps//:io_micrometer_micrometer_registry_prometheus", "@nv_third_party_deps//:io_micrometer_micrometer_tracing", diff --git a/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java b/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java index 041d2f72be..941caac736 100644 --- a/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java +++ b/src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/service/eventledger/EventLedgerClient.java @@ -24,9 +24,6 @@ import io.cloudevents.core.builder.CloudEventBuilder; import io.cloudevents.core.format.EventFormat; import io.cloudevents.core.provider.EventFormatProvider; -import io.micrometer.context.ContextSnapshot; -import io.micrometer.context.ContextSnapshotFactory; -import jakarta.annotation.PreDestroy; import java.net.URI; import java.nio.charset.StandardCharsets; import java.time.Duration; @@ -34,11 +31,6 @@ import java.time.ZoneOffset; import java.util.Map; import java.util.UUID; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; @@ -60,11 +52,6 @@ public class EventLedgerClient { private static final String MESG_UNKNOWN_FUNCTION_STATUS = "Event Ledger unknown function status: {}, skip publishing."; - private static final String MESG_QUEUE_FULL = - "Event Ledger queue is full: accountId={}, functionId={}, deploymentId={}, status={}"; - private static final String MESG_FAILED_TO_ENQUEUE_EVENT = - "Failed to enqueue Event Ledger event: accountId={}, functionId={}, deploymentId={}, " - + "status={}"; private static final String MESG_FAILED_TO_PUBLISH_FUNCTION_STATUS = "Failed to publish function status to Event Ledger: accountId={}, functionId={}, " + "functionVersionId={}, deploymentId={}, status={}"; @@ -75,14 +62,10 @@ public class EventLedgerClient { FunctionStatus.DEGRADED, "Function.Degraded", FunctionStatus.ERROR, "Function.Error", FunctionStatus.INACTIVE, "Function.Inactive"); - private static final ContextSnapshotFactory CONTEXT_SNAPSHOT_FACTORY = - ContextSnapshotFactory.builder().build(); - private final Duration timeout; private final boolean enabled; private final WebClient webClient; private final JsonMapper jsonMapper; - private final ExecutorService executor; private final EventFormat eventFormat; private record FunctionStatusTransition( @@ -100,8 +83,6 @@ public EventLedgerClient( @Value("${nvcf.event-ledger.base-url:http://event-ledger.nvcf.svc.cluster.local:8080}") String baseUrl, @Value("${nvcf.event-ledger.timeout:2s}") Duration timeout, - @Value("${nvcf.event-ledger.publisher-threads:2}") int publisherThreads, - @Value("${nvcf.event-ledger.queue-capacity:1000}") int queueCapacity, @Value("${spring.security.oauth2.client.registration.event-ledger.client-id:}") String clientId, @Value("${spring.security.oauth2.client.registration.event-ledger.client-secret:}") @@ -113,25 +94,18 @@ public EventLedgerClient( this(enabled, timeout, enabled ? authenticatedWebClient( baseUrl, clientId, clientSecret, scope, tokenUri, webClientBuilder) - : webClientBuilder.baseUrl(baseUrl).build(), jsonMapper, - new ThreadPoolExecutor(publisherThreads, publisherThreads, 0L, TimeUnit.MILLISECONDS, - new ArrayBlockingQueue<>(queueCapacity), - Thread.ofPlatform().name("event-ledger-publisher-", 0) - .factory(), - new ThreadPoolExecutor.AbortPolicy())); + : webClientBuilder.baseUrl(baseUrl).build(), jsonMapper); } EventLedgerClient( boolean enabled, Duration timeout, WebClient webClient, - JsonMapper jsonMapper, - ExecutorService executor) { + JsonMapper jsonMapper) { this.enabled = enabled; this.timeout = timeout; this.webClient = webClient; this.jsonMapper = jsonMapper; - this.executor = executor; this.eventFormat = EventFormatProvider.getInstance().resolveFormat(CONTENT_TYPE); } @@ -160,19 +134,7 @@ public void publish( return; } - try { - // Preserve tracing and logging context when publishing on the executor thread. - ContextSnapshot contextSnapshot = CONTEXT_SNAPSHOT_FACTORY.captureAll(); - executor.execute(contextSnapshot.wrap(() -> send(ncaId, transition, eventName))); - } catch (RejectedExecutionException ex) { - log.warn(MESG_QUEUE_FULL, - ncaId, transition.functionId(), transition.deploymentId(), - transition.currentStatus()); - } catch (RuntimeException ex) { - log.warn(MESG_FAILED_TO_ENQUEUE_EVENT, - ncaId, transition.functionId(), transition.deploymentId(), - transition.currentStatus(), ex); - } + send(ncaId, transition, eventName); } private void send( @@ -223,9 +185,4 @@ private static WebClient authenticatedWebClient( builder, CLIENT_REGISTRATION_ID, tokenUri, clientId, clientSecret, scope)) .build(); } - - @PreDestroy - void close() { - executor.shutdownNow(); - } } diff --git a/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java b/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java index b675040590..d9e906f4c9 100644 --- a/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java +++ b/src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/service/eventledger/EventLedgerClientTest.java @@ -23,7 +23,6 @@ import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.options; import static io.cloudevents.jackson.JsonFormat.CONTENT_TYPE; import static org.assertj.core.api.Assertions.assertThat; -import static org.awaitility.Awaitility.await; import static org.junit.jupiter.params.provider.Arguments.arguments; import com.github.tomakehurst.wiremock.WireMockServer; @@ -33,9 +32,6 @@ import java.time.Instant; import java.time.ZoneOffset; import java.util.UUID; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; import java.util.stream.Stream; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -64,7 +60,6 @@ void setUp() { @AfterEach void tearDown() { - client.close(); server.stop(); } @@ -85,7 +80,7 @@ void publishesStructuredCloudEvent(FunctionStatus status, String eventName) thro "account-1", functionId, functionVersionId, deploymentId, previousStatus, status, persistedAt); - await().untilAsserted(() -> assertThat(server.getAllServeEvents()).hasSize(1)); + assertThat(server.getAllServeEvents()).hasSize(1); var request = server.getAllServeEvents().getFirst().getRequest(); var cloudEvent = EventFormatProvider.getInstance() .resolveFormat(CONTENT_TYPE) @@ -158,12 +153,24 @@ void handlesPublishFailureWithoutThrowing() { publish(FunctionStatus.DEPLOYING, FunctionStatus.ERROR); - await().untilAsserted(() -> assertThat(server.getAllServeEvents()).hasSize(1)); + assertThat(server.getAllServeEvents()).hasSize(1); + } + + @Test + void waitsForEventLedgerResponse() { + server.stubFor(post(urlEqualTo(EventLedgerClient.CLOUD_EVENTS_PATH)) + .willReturn(aResponse().withStatus(202).withFixedDelay(250))); + + var startedAt = Instant.now(); + publish(FunctionStatus.DEPLOYING, FunctionStatus.ACTIVE); + + assertThat(Duration.between(startedAt, Instant.now())) + .isGreaterThanOrEqualTo(Duration.ofMillis(200)); + assertThat(server.getAllServeEvents()).hasSize(1); } @Test void skipsPublishingWhenDisabled() { - client.close(); client = newClient(Duration.ofSeconds(1), false); publish(FunctionStatus.DEPLOYING, FunctionStatus.ACTIVE); @@ -176,14 +183,11 @@ private EventLedgerClient newClient(Duration timeout) { } private EventLedgerClient newClient(Duration timeout, boolean enabled) { - var executor = new ThreadPoolExecutor( - 1, 1, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(10)); return new EventLedgerClient( enabled, timeout, WebClient.builder().baseUrl(server.baseUrl()).build(), - new JsonMapper(), - executor); + new JsonMapper()); } private void publish(