CAMEL-20919: camel-ftp - Add producer health check

Author: davsclaus

components/camel-ftp/pom.xml

@@ -40,6 +40,10 @@
             <groupId>org.apache.camel</groupId>
             <artifactId>camel-file</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-health</artifactId>
+        </dependency>
         <!-- needed for dynamic to -->
         <dependency>
             <groupId>org.apache.camel</groupId>

components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileEndpoint.java

@@ -93,6 +93,11 @@ protected RemoteFileEndpoint(String uri, RemoteFileComponent<T> component, Remot
         setPollStrategy(new RemoteFilePollingConsumerPollStrategy());
     }
 
+    @Override
+    public RemoteFileComponent getComponent() {
+        return (RemoteFileComponent) super.getComponent();
+    }
+
     @Override
     public boolean isSingletonProducer() {
         // this producer is stateful because the remote file operations is not

components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileProducer.java

@@ -20,6 +20,8 @@
 import org.apache.camel.ExchangePropertyKey;
 import org.apache.camel.component.file.GenericFileOperationFailedException;
 import org.apache.camel.component.file.GenericFileProducer;
+import org.apache.camel.health.HealthCheckHelper;
+import org.apache.camel.health.WritableHealthCheckRepository;
 import org.apache.camel.util.URISupport;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -31,6 +33,8 @@ public class RemoteFileProducer<T> extends GenericFileProducer<T> {
 
     private static final Logger LOG = LoggerFactory.getLogger(RemoteFileProducer.class);
     private boolean loggedIn;
+    private RemoteFileProducerHealthCheck producerHealthCheck;
+    private WritableHealthCheckRepository healthCheckRepository;
 
     private transient String remoteFileProducerToString;
 
@@ -105,11 +109,14 @@ public void disconnect() throws GenericFileOperationFailedException {
 
     @Override
     public void preWriteCheck(Exchange exchange) throws Exception {
-        // before writing send a noop to see if the connection is alive and
-        // works
+        doPreWriteCheck(exchange, getEndpoint().getConfiguration().isSendNoop());
+    }
+
+    protected void doPreWriteCheck(Exchange exchange, boolean sendNoop) throws Exception {
+        // before writing send a noop to see if the connection is alive and works
         boolean noop = false;
         if (loggedIn) {
-            if (getEndpoint().getConfiguration().isSendNoop()) {
+            if (sendNoop) {
                 try {
                     noop = getOperations().sendNoop();
                 } catch (Exception e) {
@@ -120,8 +127,7 @@ public void preWriteCheck(Exchange exchange) throws Exception {
                 }
                 LOG.trace("preWriteCheck send noop success: {}", noop);
             } else {
-                // okay send noop is disabled then we would regard the op as
-                // success
+                // okay send noop is disabled then we would regard the op as success
                 noop = true;
                 LOG.trace("preWriteCheck send noop disabled");
             }
@@ -162,14 +168,30 @@ public void postWriteCheck(Exchange exchange) {
     @Override
     protected void doStart() throws Exception {
         LOG.debug("Starting");
+
+        // health-check is optional so discover and resolve
+        healthCheckRepository = HealthCheckHelper.getHealthCheckRepository(
+                endpoint.getCamelContext(),
+                "producers",
+                WritableHealthCheckRepository.class);
+
+        if (healthCheckRepository != null) {
+            producerHealthCheck = new RemoteFileProducerHealthCheck(this);
+            producerHealthCheck.setEnabled(this.getEndpoint().getComponent().isHealthCheckProducerEnabled());
+            healthCheckRepository.addHealthCheck(producerHealthCheck);
+        }
+
         // do not connect when component starts, just wait until we process as
-        // we will
-        // connect at that time if needed
+        // we will connect at that time if needed
         super.doStart();
     }
 
     @Override
     protected void doStop() throws Exception {
+        if (healthCheckRepository != null && producerHealthCheck != null) {
+            healthCheckRepository.removeHealthCheck(producerHealthCheck);
+            producerHealthCheck = null;
+        }
         try {
             disconnect();
         } catch (Exception e) {

components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileProducerHealthCheck.java

@@ -0,0 +1,64 @@
+/*
+ * 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.file.remote;
+
+import java.util.Map;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.component.file.GenericFileOperationFailedException;
+import org.apache.camel.health.HealthCheckResultBuilder;
+import org.apache.camel.impl.health.AbstractHealthCheck;
+
+/**
+ * FTP producer readiness health-check
+ */
+public class RemoteFileProducerHealthCheck extends AbstractHealthCheck {
+
+    private final RemoteFileProducer<?> producer;
+
+    public RemoteFileProducerHealthCheck(RemoteFileProducer<?> producer) {
+        super("camel", "producer:ftp-" + producer.getEndpoint().getConfiguration().getHost());
+        this.producer = producer;
+    }
+
+    @Override
+    protected void doCall(HealthCheckResultBuilder builder, Map<String, Object> options) {
+        Exchange dummy = producer.createExchange();
+        Exception cause = null;
+        try {
+            producer.doPreWriteCheck(dummy, true);
+        } catch (Exception e) {
+            cause = e;
+        }
+        if (cause != null) {
+            builder.down();
+            builder.message("FtpProducer is not ready");
+            builder.detail("serviceUrl", producer.getEndpoint().getServiceUrl());
+            builder.error(cause);
+            if (cause instanceof GenericFileOperationFailedException gfe) {
+                int code = gfe.getCode();
+                String msg = gfe.getReason();
+                if (code > 0 && msg != null) {
+                    builder.detail("ftp.code", code);
+                    builder.detail("ftp.reason", msg.trim());
+                }
+            }
+        } else {
+            builder.up();
+        }
+    }
+}

components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/integration/FtpProducerHealthCheckIT.java

@@ -0,0 +1,92 @@
+/*
+ * 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.file.remote.integration;
+
+import java.util.Collection;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.file.remote.FtpConstants;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.health.HealthCheck;
+import org.apache.camel.health.HealthCheckHelper;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+
+public class FtpProducerHealthCheckIT extends FtpServerTestSupport {
+
+    private String getFtpUrl() {
+        return "ftp://admin@localhost:{{ftp.server.port}}/reply?password=admin";
+    }
+
+    @Override
+    protected CamelContext createCamelContext() throws Exception {
+        CamelContext context = super.createCamelContext();
+        HealthCheckHelper.getHealthCheckRepository(context, "producers").setEnabled(true);
+        return context;
+    }
+
+    @Test
+    public void testHealthCheck() throws Exception {
+        MockEndpoint mock = getMockEndpoint("mock:result");
+        mock.expectedBodiesReceived("Bye World");
+        mock.expectedHeaderReceived(FtpConstants.FTP_REPLY_CODE, 226);
+        mock.expectedHeaderReceived(FtpConstants.FTP_REPLY_STRING, "226 Transfer complete.");
+
+        template.requestBodyAndHeader("direct:start", "Bye World", Exchange.FILE_NAME, "hello.txt");
+
+        MockEndpoint.assertIsSatisfied(context);
+
+        await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
+            Collection<HealthCheck.Result> res = HealthCheckHelper.invokeReadiness(context);
+            boolean up = res.stream().allMatch(r -> r.getState().equals(HealthCheck.State.UP));
+            Assertions.assertTrue(up, "readiness check");
+        });
+
+        // stop FTP server
+        service.shutdown();
+
+        // health-check should then become down
+        await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
+            Collection<HealthCheck.Result> res = HealthCheckHelper.invokeReadiness(context);
+            Optional<HealthCheck.Result> hr = res.stream().filter(r -> r.getState().equals(HealthCheck.State.DOWN)).findFirst();
+            Assertions.assertTrue(hr.isPresent());
+            HealthCheck.Result r = hr.get();
+            Assertions.assertEquals(HealthCheck.State.DOWN, r.getState());
+            Assertions.assertEquals("FtpProducer is not ready", r.getMessage().get());
+            Assertions.assertEquals(200, r.getDetails().get("ftp.code"));
+            Assertions.assertEquals("Connection refused", r.getDetails().get("ftp.reason"));
+        });
+
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            @Override
+            public void configure() {
+                from("direct:start").to(getFtpUrl()).to("mock:result");
+            }
+        };
+    }
+
+}