Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
/**
* Copyright (C) 2006-2026 Talend Inc. - www.talend.com
*
* 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 org.talend.sdk.component.server.api;

import static javax.ws.rs.core.MediaType.APPLICATION_JSON;

import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.core.Response;

import org.eclipse.microprofile.openapi.annotations.Operation;
import org.eclipse.microprofile.openapi.annotations.media.Content;
import org.eclipse.microprofile.openapi.annotations.media.Schema;
import org.eclipse.microprofile.openapi.annotations.responses.APIResponse;
import org.eclipse.microprofile.openapi.annotations.tags.Tag;
import org.talend.sdk.component.server.front.model.HealthStatus;

@Path("liveness")
@Tag(name = "Health", description = "Kubernetes liveness probe endpoint.")
Comment thread
undx marked this conversation as resolved.
public interface LivenessResource {

@GET
@Produces(APPLICATION_JSON)
@Operation(operationId = "getLiveness",
description = "Liveness probe: returns 200 when the JVM is healthy (no fatal error recorded). "
+ "Returns 503 with a cause when a VirtualMachineError (e.g. OutOfMemoryError) has been intercepted.")
@APIResponse(responseCode = "200",
description = "Application is healthy.",
content = @Content(mediaType = APPLICATION_JSON,
schema = @Schema(implementation = HealthStatus.class)))
@APIResponse(responseCode = "503",
description = "Application has encountered a fatal error.",
content = @Content(mediaType = APPLICATION_JSON,
schema = @Schema(implementation = HealthStatus.class)))
Response getLiveness();
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
/**
* Copyright (C) 2006-2026 Talend Inc. - www.talend.com
*
* 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 org.talend.sdk.component.server.api;

import static javax.ws.rs.core.MediaType.APPLICATION_JSON;

import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.core.Response;

import org.eclipse.microprofile.openapi.annotations.Operation;
import org.eclipse.microprofile.openapi.annotations.media.Content;
import org.eclipse.microprofile.openapi.annotations.media.Schema;
import org.eclipse.microprofile.openapi.annotations.responses.APIResponse;
import org.eclipse.microprofile.openapi.annotations.tags.Tag;
import org.talend.sdk.component.server.front.model.HealthStatus;

@Path("readiness")
@Tag(name = "Health", description = "Kubernetes readiness probe endpoint.")
public interface ReadinessResource {

@GET
@Produces(APPLICATION_JSON)
@Operation(operationId = "getReadiness",
description = "Readiness probe: returns 200 when the component index is loaded and the server is ready "
+ "to serve traffic. Returns 503 with a cause otherwise.")
@APIResponse(responseCode = "200",
description = "Server is ready.",
content = @Content(mediaType = APPLICATION_JSON,
schema = @Schema(implementation = HealthStatus.class)))
@APIResponse(responseCode = "503",
description = "Server is not ready.",
content = @Content(mediaType = APPLICATION_JSON,
schema = @Schema(implementation = HealthStatus.class)))
Response getReadiness();
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
/**
* Copyright (C) 2006-2026 Talend Inc. - www.talend.com
*
* 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 org.talend.sdk.component.server.front.model;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@AllArgsConstructor
@NoArgsConstructor
public class HealthStatus {

private String status;

private String cause;
}
6 changes: 6 additions & 0 deletions component-server-parent/component-server/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -158,6 +158,12 @@
<version>${meecrowave.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>${mockito4.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.tomee</groupId>
<artifactId>ziplock</artifactId>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -201,6 +201,12 @@ public class ComponentServerConfiguration {
@ConfigProperty(name = "talend.component.server.plugins.reloading.marker")
private Optional<String> pluginsReloadFileMarker;

@Inject
@Documentation("Whether the Vault connectivity check is included in the readiness probe. "
+ "Set to true only when this server instance uses Vault for credential decryption.")
@ConfigProperty(name = "talend.server.health.vault.enabled", defaultValue = "false")
private Boolean healthVaultEnabled;

Comment thread
undx marked this conversation as resolved.
@PostConstruct
private void init() {
if (logRequests != null && logRequests) {
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
/**
* Copyright (C) 2006-2026 Talend Inc. - www.talend.com
*
* 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 org.talend.sdk.component.server.front;

import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import javax.ws.rs.core.Response;

import org.talend.sdk.component.server.api.LivenessResource;
import org.talend.sdk.component.server.front.model.HealthStatus;
import org.talend.sdk.component.server.service.FatalState;

@ApplicationScoped
public class LivenessResourceImpl implements LivenessResource {

private static final String STATUS_UP = "UP";

private static final String STATUS_DOWN = "DOWN";

@Inject
private FatalState fatalState;

@Override
public Response getLiveness() {
if (fatalState.hasFatalError()) {
return Response
.status(Response.Status.SERVICE_UNAVAILABLE)
.entity(new HealthStatus(STATUS_DOWN, fatalState.getCause()))
.build();
}
return Response.ok(new HealthStatus(STATUS_UP, null)).build();
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
/**
* Copyright (C) 2006-2026 Talend Inc. - www.talend.com
*
* 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 org.talend.sdk.component.server.front;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import javax.ws.rs.core.Response;

import org.talend.sdk.component.server.api.ReadinessResource;
import org.talend.sdk.component.server.configuration.ComponentServerConfiguration;
import org.talend.sdk.component.server.front.model.HealthStatus;
import org.talend.sdk.component.server.service.ComponentManagerService;
import org.talend.sdk.components.vault.client.VaultClient;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@ApplicationScoped
public class ReadinessResourceImpl implements ReadinessResource {

private static final String STATUS_UP = "UP";

private static final String STATUS_DOWN = "DOWN";

private static final long VAULT_CACHE_TTL_MS = 5_000L;

@Inject
private ComponentManagerService componentManagerService;

@Inject
private ComponentServerConfiguration configuration;

@Inject
private VaultClient vaultClient;

private final AtomicBoolean cachedVaultResult = new AtomicBoolean(true);

private final AtomicLong lastVaultCheck = new AtomicLong(0L);

@Override
public Response getReadiness() {
if (!componentManagerService.isStarted()) {
return Response
.status(Response.Status.SERVICE_UNAVAILABLE)
.entity(new HealthStatus(STATUS_DOWN, "Component index not ready"))
.build();
}
if (configuration.getHealthVaultEnabled()) {
final HealthStatus vaultStatus = checkVaultCached();
if (STATUS_DOWN.equals(vaultStatus.getStatus())) {
return Response.status(Response.Status.SERVICE_UNAVAILABLE).entity(vaultStatus).build();
}
}
return Response.ok(new HealthStatus(STATUS_UP, null)).build();
}

private HealthStatus checkVaultCached() {
final long now = System.currentTimeMillis();
if (now - lastVaultCheck.get() < VAULT_CACHE_TTL_MS) {
return cachedVaultResult.get()
? new HealthStatus(STATUS_UP, null)
: new HealthStatus(STATUS_DOWN, "Vault is not reachable");
}
try {
final boolean reachable = vaultClient.ping();
cachedVaultResult.set(reachable);
lastVaultCheck.set(now);
if (!reachable) {
log.warn("Readiness check: Vault is not reachable");
return new HealthStatus(STATUS_DOWN, "Vault is not reachable");
}
} catch (final VirtualMachineError vme) {
throw vme;
} catch (final Throwable t) {
cachedVaultResult.set(false);
lastVaultCheck.set(now);
final String cause = "Vault connectivity check failed: " + t.getMessage();
log.warn("Readiness check failed: {}", cause);
return new HealthStatus(STATUS_DOWN, cause);
}
return new HealthStatus(STATUS_UP, null);
}
Comment thread
undx marked this conversation as resolved.
Comment on lines +74 to +98
}
Original file line number Diff line number Diff line change
Expand Up @@ -22,14 +22,17 @@
import javax.annotation.PostConstruct;
import javax.enterprise.context.Dependent;
import javax.inject.Inject;
import javax.servlet.http.HttpServletRequest;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.Response;
import javax.ws.rs.ext.ExceptionMapper;
import javax.ws.rs.ext.Provider;

import org.talend.sdk.component.server.configuration.ComponentServerConfiguration;
import org.talend.sdk.component.server.front.model.ErrorDictionary;
import org.talend.sdk.component.server.front.model.error.ErrorPayload;
import org.talend.sdk.component.server.service.FatalState;

import lombok.extern.slf4j.Slf4j;

Expand All @@ -41,6 +44,12 @@ public class DefaultExceptionHandler implements ExceptionMapper<Throwable> {
@Inject
private ComponentServerConfiguration configuration;

@Inject
private FatalState fatalState;

@Context
private HttpServletRequest request;

private boolean replaceException;

@PostConstruct
Expand All @@ -50,6 +59,12 @@ private void init() {

@Override
public Response toResponse(final Throwable exception) {
if (exception instanceof VirtualMachineError) {
final String requestPath = request != null ? request.getRequestURI() : "(unknown)";
final String cause = exception.getClass().getSimpleName() + " during request "
+ requestPath + ": " + exception.getMessage();
fatalState.markFatal(cause);
Comment thread
undx marked this conversation as resolved.
}
Comment on lines +62 to +67
log.error("[DefaultExceptionHandler#toResponse] Throwable: ", exception);
final Response response;
if (exception instanceof WebApplicationException) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,7 @@ public class ComponentManagerService {

private Connectors connectors;

private boolean started;
private volatile boolean started;

private Path m2;

Expand Down Expand Up @@ -460,4 +460,8 @@ public ComponentManager manager() {
return instance;
}

public boolean isStarted() {
return started;
}
Comment thread
undx marked this conversation as resolved.

}
Loading
Loading