This is an automated email from the ASF dual-hosted git repository. acosentino pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel.git
commit efb0cfab8352832424c3638e187aa460d47aa37f Author: Andrea Cosentino <[email protected]> AuthorDate: Wed Nov 9 11:37:57 2022 +0100 CAMEL-18131 - camel-health - Add health checks for components that has extension for connectivity verification - DynamoDB Streams Signed-off-by: Andrea Cosentino <[email protected]> --- .../aws2/ddbstream/Ddb2StreamComponent.java | 2 - .../Ddb2StreamComponentVerifierExtension.java | 94 --------------------- .../aws2/ddbstream/Ddb2StreamConsumer.java | 21 +++++ .../ddbstream/Ddb2StreamConsumerHealthCheck.java | 96 ++++++++++++++++++++++ .../DdbStreamComponentVerifierExtensionTest.java | 90 -------------------- 5 files changed, 117 insertions(+), 186 deletions(-) diff --git a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamComponent.java b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamComponent.java index dab1869eb4f..ea9c73850bf 100644 --- a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamComponent.java +++ b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamComponent.java @@ -40,8 +40,6 @@ public class Ddb2StreamComponent extends DefaultComponent { public Ddb2StreamComponent(CamelContext context) { super(context); - - registerExtension(new Ddb2StreamComponentVerifierExtension()); } @Override diff --git a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamComponentVerifierExtension.java b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamComponentVerifierExtension.java deleted file mode 100644 index e7da3cac1c8..00000000000 --- a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamComponentVerifierExtension.java +++ /dev/null @@ -1,94 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF 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 - * - * 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.apache.camel.component.aws2.ddbstream; - -import java.util.Map; - -import org.apache.camel.component.extension.verifier.DefaultComponentVerifierExtension; -import org.apache.camel.component.extension.verifier.ResultBuilder; -import org.apache.camel.component.extension.verifier.ResultErrorBuilder; -import org.apache.camel.component.extension.verifier.ResultErrorHelper; -import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; -import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; -import software.amazon.awssdk.core.exception.SdkClientException; -import software.amazon.awssdk.regions.Region; -import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsClient; -import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsClientBuilder; - -public class Ddb2StreamComponentVerifierExtension extends DefaultComponentVerifierExtension { - - public Ddb2StreamComponentVerifierExtension() { - this("aws2-ddbstream"); - } - - public Ddb2StreamComponentVerifierExtension(String scheme) { - super(scheme); - } - - // ********************************* - // Parameters validation - // ********************************* - - @Override - protected Result verifyParameters(Map<String, Object> parameters) { - - ResultBuilder builder = ResultBuilder.withStatusAndScope(Result.Status.OK, Scope.PARAMETERS) - .error(ResultErrorHelper.requiresOption("accessKey", parameters)) - .error(ResultErrorHelper.requiresOption("secretKey", parameters)) - .error(ResultErrorHelper.requiresOption("region", parameters)); - - // Validate using the catalog - - super.verifyParametersAgainstCatalog(builder, parameters); - - return builder.build(); - } - - // ********************************* - // Connectivity validation - // ********************************* - - @Override - protected Result verifyConnectivity(Map<String, Object> parameters) { - ResultBuilder builder = ResultBuilder.withStatusAndScope(Result.Status.OK, Scope.CONNECTIVITY); - - try { - Ddb2StreamConfiguration configuration = setProperties(new Ddb2StreamConfiguration(), parameters); - if (!DynamoDbStreamsClient.serviceMetadata().regions().contains(Region.of(configuration.getRegion()))) { - ResultErrorBuilder errorBuilder = ResultErrorBuilder.withCodeAndDescription( - VerificationError.StandardCode.ILLEGAL_PARAMETER, "The service is not supported in this region"); - return builder.error(errorBuilder.build()).build(); - } - AwsBasicCredentials cred = AwsBasicCredentials.create(configuration.getAccessKey(), configuration.getSecretKey()); - DynamoDbStreamsClientBuilder clientBuilder = DynamoDbStreamsClient.builder(); - DynamoDbStreamsClient client = clientBuilder.credentialsProvider(StaticCredentialsProvider.create(cred)) - .region(Region.of(configuration.getRegion())).build(); - client.listStreams(); - } catch (SdkClientException e) { - ResultErrorBuilder errorBuilder - = ResultErrorBuilder.withCodeAndDescription(VerificationError.StandardCode.AUTHENTICATION, e.getMessage()) - .detail("aws_ddb_stream_exception_message", e.getMessage()) - .detail(VerificationError.ExceptionAttribute.EXCEPTION_CLASS, e.getClass().getName()) - .detail(VerificationError.ExceptionAttribute.EXCEPTION_INSTANCE, e); - - builder.error(errorBuilder.build()); - } catch (Exception e) { - builder.error(ResultErrorBuilder.withException(e).build()); - } - return builder.build(); - } -} diff --git a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java index 25044fc7b87..5b9d6c6fdea 100644 --- a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java +++ b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java @@ -26,6 +26,8 @@ import java.util.Queue; import org.apache.camel.AsyncCallback; import org.apache.camel.Exchange; import org.apache.camel.Processor; +import org.apache.camel.health.HealthCheckHelper; +import org.apache.camel.health.WritableHealthCheckRepository; import org.apache.camel.support.ScheduledBatchPollingConsumer; import org.apache.camel.util.CastUtils; import org.apache.camel.util.ObjectHelper; @@ -43,6 +45,10 @@ public class Ddb2StreamConsumer extends ScheduledBatchPollingConsumer { private final ShardIteratorHandler shardIteratorHandler; private final Map<String, String> lastSeenSequenceNumbers = new HashMap<>(); + private WritableHealthCheckRepository healthCheckRepository; + + private Ddb2StreamConsumerHealthCheck consumerHealthCheck; + public Ddb2StreamConsumer(Ddb2StreamEndpoint endpoint, Processor processor) { this(endpoint, processor, new ShardIteratorHandler(endpoint)); } @@ -106,6 +112,21 @@ public class Ddb2StreamConsumer extends ScheduledBatchPollingConsumer { return processedExchanges; } + @Override + protected void doStart() throws Exception { + super.doStart(); + + healthCheckRepository = HealthCheckHelper.getHealthCheckRepository( + getEndpoint().getCamelContext(), + "components", + WritableHealthCheckRepository.class); + + if (healthCheckRepository != null) { + consumerHealthCheck = new Ddb2StreamConsumerHealthCheck(this, getRouteId()); + healthCheckRepository.addHealthCheck(consumerHealthCheck); + } + } + protected Exchange createExchange(Record record) { Exchange ex = createExchange(true); ex.getIn().setBody(record, Record.class); diff --git a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumerHealthCheck.java b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumerHealthCheck.java new file mode 100644 index 00000000000..9357f015b55 --- /dev/null +++ b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumerHealthCheck.java @@ -0,0 +1,96 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF 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 + * + * 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.apache.camel.component.aws2.ddbstream; + +import org.apache.camel.health.HealthCheckResultBuilder; +import org.apache.camel.impl.health.AbstractHealthCheck; +import org.apache.camel.util.ObjectHelper; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.awscore.exception.AwsServiceException; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsClient; +import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsClientBuilder; + +import java.util.Map; + +public class Ddb2StreamConsumerHealthCheck extends AbstractHealthCheck { + + private final Ddb2StreamConsumer ddb2StreamConsumer; + private final String routeId; + + public Ddb2StreamConsumerHealthCheck(Ddb2StreamConsumer ddb2StreamConsumer, String routeId) { + super("camel", "aws2-dd2-stream-consumer-" + routeId); + this.ddb2StreamConsumer = ddb2StreamConsumer; + this.routeId = routeId; + } + + @Override + public boolean isLiveness() { + // this health check is only readiness + return false; + } + + @Override + protected void doCall(HealthCheckResultBuilder builder, Map<String, Object> options) { + + try { + Ddb2StreamConfiguration configuration = ddb2StreamConsumer.getEndpoint().getConfiguration(); + if (ObjectHelper.isNotEmpty(configuration.getRegion())) { + if (!DynamoDbStreamsClient.serviceMetadata().regions().contains(Region.of(configuration.getRegion()))) { + builder.message("The service is not supported in this region"); + builder.down(); + return; + } + } + DynamoDbStreamsClient client; + if (!configuration.isUseDefaultCredentialsProvider()) { + AwsBasicCredentials cred + = AwsBasicCredentials.create(configuration.getAccessKey(), configuration.getSecretKey()); + DynamoDbStreamsClientBuilder clientBuilder = DynamoDbStreamsClient.builder(); + client = clientBuilder.credentialsProvider(StaticCredentialsProvider.create(cred)) + .region(Region.of(configuration.getRegion())).build(); + } else if (ObjectHelper.isNotEmpty(configuration.getAmazonDynamoDbStreamsClient())) { + client = configuration.getAmazonDynamoDbStreamsClient(); + } else { + DynamoDbStreamsClientBuilder clientBuilder = DynamoDbStreamsClient.builder(); + client = clientBuilder.region(Region.of(configuration.getRegion())).build(); + } + client.listStreams(); + } catch (AwsServiceException e) { + builder.message(e.getMessage()); + builder.error(e); + if (ObjectHelper.isNotEmpty(e.statusCode())) { + builder.detail(SERVICE_STATUS_CODE, e.statusCode()); + } + if (ObjectHelper.isNotEmpty(e.awsErrorDetails().errorCode())) { + builder.detail(SERVICE_ERROR_CODE, e.awsErrorDetails().errorCode()); + } + builder.down(); + return; + + } catch (Exception e) { + builder.error(e); + builder.down(); + return; + } + + builder.up(); + + } + +} diff --git a/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/DdbStreamComponentVerifierExtensionTest.java b/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/DdbStreamComponentVerifierExtensionTest.java deleted file mode 100644 index 81c4510452c..00000000000 --- a/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/DdbStreamComponentVerifierExtensionTest.java +++ /dev/null @@ -1,90 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF 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 - * - * 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.apache.camel.component.aws2.ddbstream; - -import java.util.HashMap; -import java.util.Map; - -import org.apache.camel.Component; -import org.apache.camel.component.extension.ComponentVerifierExtension; -import org.apache.camel.test.junit5.CamelTestSupport; -import org.junit.jupiter.api.Test; - -import static org.junit.jupiter.api.Assertions.assertEquals; - -public class DdbStreamComponentVerifierExtensionTest extends CamelTestSupport { - - // ************************************************* - // Tests (parameters) - // ************************************************* - @Override - public boolean isUseRouteBuilder() { - return false; - } - - @Test - public void testParameters() { - Component component = context().getComponent("aws2-ddbstream"); - - ComponentVerifierExtension verifier - = component.getExtension(ComponentVerifierExtension.class).orElseThrow(IllegalStateException::new); - - Map<String, Object> parameters = new HashMap<>(); - parameters.put("secretKey", "l"); - parameters.put("accessKey", "k"); - parameters.put("region", "l"); - parameters.put("tableName", "test"); - - ComponentVerifierExtension.Result result = verifier.verify(ComponentVerifierExtension.Scope.PARAMETERS, parameters); - - assertEquals(ComponentVerifierExtension.Result.Status.OK, result.getStatus()); - } - - @Test - public void testConnectivity() { - Component component = context().getComponent("aws2-ddbstream"); - ComponentVerifierExtension verifier - = component.getExtension(ComponentVerifierExtension.class).orElseThrow(IllegalStateException::new); - - Map<String, Object> parameters = new HashMap<>(); - parameters.put("secretKey", "l"); - parameters.put("accessKey", "k"); - parameters.put("region", "US_EAST_1"); - parameters.put("tableName", "test"); - - ComponentVerifierExtension.Result result = verifier.verify(ComponentVerifierExtension.Scope.CONNECTIVITY, parameters); - - assertEquals(ComponentVerifierExtension.Result.Status.ERROR, result.getStatus()); - } - - @Test - public void testConnectivityAndRegion() { - Component component = context().getComponent("aws2-ddbstream"); - ComponentVerifierExtension verifier - = component.getExtension(ComponentVerifierExtension.class).orElseThrow(IllegalStateException::new); - - Map<String, Object> parameters = new HashMap<>(); - parameters.put("secretKey", "l"); - parameters.put("accessKey", "k"); - parameters.put("region", "l"); - parameters.put("tableName", "test"); - - ComponentVerifierExtension.Result result = verifier.verify(ComponentVerifierExtension.Scope.CONNECTIVITY, parameters); - - assertEquals(ComponentVerifierExtension.Result.Status.ERROR, result.getStatus()); - } -}
