KNOX-2955 - Knox Readiness Awareness and Notification

gateway-server/src/main/java/org/apache/knox/gateway/GatewayMessages.java

@@ -18,6 +18,7 @@
 package org.apache.knox.gateway;
 
 import java.io.File;
+import java.io.IOException;
 import java.net.URI;
 import java.time.Instant;
 import java.util.Date;
@@ -32,8 +33,6 @@
 import org.apache.knox.gateway.services.security.KeystoreServiceException;
 import org.apache.knox.gateway.topology.monitor.db.LocalDirectory;
 
-import java.io.IOException;
-
 @Messages(logger="org.apache.knox.gateway")
 public interface GatewayMessages {
 
@@ -785,4 +784,20 @@ void failedToDiscoverClusterServices(String clusterName, String topologyName,
   @Message(level = MessageLevel.DEBUG,
           text = "Request {0} is wrapped to url encoded form request.")
   void wrappingRequestToUrlEncodedFormRequest(String requestURI);
+
+  @Message(level = MessageLevel.INFO,
+          text = "Checking gateway status. Deployed topologies: {0}. Waiting for: {1}")
+  void checkingGatewayStatus(Set<String> deployedTopologies, Set<String> missingTopologies);
+
+  @Message(level = MessageLevel.INFO,
+          text = "Checking gateway status. No topologies to check")
+  void noTopologiesToCheck();
+
+  @Message(level = MessageLevel.INFO,
+          text = "Collected topologies for health check: {0}")
+  void collectedTopologiesForHealthCheck(Set<String> result);
+
+  @Message(level = MessageLevel.INFO,
+          text = "Starting gateway status service. Topologies to check: {0}")
+  void startingStatusMonitor(Set<String> topologyNames);
 }

gateway-server/src/main/java/org/apache/knox/gateway/GatewayServer.java

@@ -43,6 +43,7 @@
 import org.apache.knox.gateway.services.security.AliasServiceException;
 import org.apache.knox.gateway.services.security.SSLService;
 import org.apache.knox.gateway.services.topology.TopologyService;
+import org.apache.knox.gateway.services.topology.impl.GatewayStatusService;
 import org.apache.knox.gateway.topology.Application;
 import org.apache.knox.gateway.topology.Topology;
 import org.apache.knox.gateway.topology.TopologyEvent;
@@ -146,6 +147,7 @@ public class GatewayServer {
   private TopologyListener listener;
   private Map<String, WebAppContext> deployments;
   private AtomicBoolean stopped = new AtomicBoolean(false);
+  private GatewayStatusService gatewayStatusService;
 
   public static void main( String[] args ) {
     try {
@@ -630,6 +632,10 @@ private synchronized void start() throws Exception {
     // by the descriptor monitor
     handleHadoopXmlResources();
 
+    // at this point descriptors are supposed to be generated from hxr
+    gatewayStatusService = services.getService(ServiceType.GATEWAY_STATUS_SERVICE);
+    gatewayStatusService.initTopologiesToCheck();
+
     monitor.addTopologyChangeListener(listener);
     log.loadingTopologiesFromDirectory(topologiesDir.getAbsolutePath());
     monitor.reloadTopologies();
@@ -1038,6 +1044,7 @@ private void handleCreateDeployment(Topology topology, File deployDir) {
           log.redeployedTopology( topology.getName() );
         }
         cleanupTopologyDeployments( deployDir, topology );
+        gatewayStatusService.onTopologyReady(topology.getName());
       } catch( Throwable e ) {
         auditor.audit( Action.DEPLOY, topology.getName(), ResourceType.TOPOLOGY, ActionOutcome.FAILURE );
         log.failedToDeployTopology( topology.getName(), e );

gateway-server/src/main/java/org/apache/knox/gateway/config/impl/GatewayConfigImpl.java

@@ -331,6 +331,8 @@ public class GatewayConfigImpl extends Configuration implements GatewayConfig {
   private static final String GATEWAY_SERVLET_ASYNC_SUPPORTED = GATEWAY_CONFIG_FILE_PREFIX + ".servlet.async.supported";
   private static final boolean GATEWAY_SERVLET_ASYNC_SUPPORTED_DEFAULT = false;
 
+  private static final String GATEWAY_HEALTH_CHECK_TOPOLOGIES = GATEWAY_CONFIG_FILE_PREFIX + ".health.check.topologies";
+
   public GatewayConfigImpl() {
     init();
   }
@@ -1469,6 +1471,12 @@ public long getConcurrentSessionVerifierExpiredTokensCleaningPeriod() {
     return getLong(GATEWAY_SESSION_VERIFICATION_EXPIRED_TOKENS_CLEANING_PERIOD, GATEWAY_SESSION_VERIFICATION_EXPIRED_TOKENS_CLEANING_PERIOD_DEFAULT);
   }
 
+  @Override
+  public Set<String> getHealthCheckTopologies() {
+    final Collection<String> topologies = getTrimmedStringCollection(GATEWAY_HEALTH_CHECK_TOPOLOGIES);
+    return topologies == null ? Collections.emptySet() : new HashSet<>(topologies);
+  }
+
   @Override
   public boolean isAsyncSupported() {
     return getBoolean(GATEWAY_SERVLET_ASYNC_SUPPORTED, GATEWAY_SERVLET_ASYNC_SUPPORTED_DEFAULT);

gateway-server/src/main/java/org/apache/knox/gateway/services/DefaultGatewayServices.java

@@ -80,6 +80,8 @@ public void init(GatewayConfig config, Map<String,String> options) throws Servic
     addService(ServiceType.METRICS_SERVICE, gatewayServiceFactory.create(this, ServiceType.METRICS_SERVICE, config, options));
 
     addService(ServiceType.CONCURRENT_SESSION_VERIFIER, gatewayServiceFactory.create(this, ServiceType.CONCURRENT_SESSION_VERIFIER, config, options));
+
+    addService(ServiceType.GATEWAY_STATUS_SERVICE, gatewayServiceFactory.create(this, ServiceType.GATEWAY_STATUS_SERVICE, config, options));
   }
 
   @Override

gateway-server/src/main/java/org/apache/knox/gateway/services/factory/GatewayStatusServiceFactory.java

@@ -0,0 +1,49 @@
+/*
+ * 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.knox.gateway.services.factory;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.unmodifiableList;
+
+import java.util.Collection;
+import java.util.Map;
+
+import org.apache.knox.gateway.config.GatewayConfig;
+import org.apache.knox.gateway.services.GatewayServices;
+import org.apache.knox.gateway.services.Service;
+import org.apache.knox.gateway.services.ServiceLifecycleException;
+import org.apache.knox.gateway.services.ServiceType;
+import org.apache.knox.gateway.services.topology.impl.GatewayStatusService;
+
+public class GatewayStatusServiceFactory extends AbstractServiceFactory {
+
+  @Override
+  protected Service createService(GatewayServices gatewayServices, ServiceType serviceType, GatewayConfig gatewayConfig, Map<String, String> options, String implementation) throws ServiceLifecycleException {
+    return new GatewayStatusService();
+  }
+
+  @Override
+  protected ServiceType getServiceType() {
+    return ServiceType.GATEWAY_STATUS_SERVICE;
+  }
+
+  @Override
+  protected Collection<String> getKnownImplementations() {
+    return unmodifiableList(asList(GatewayStatusService.class.getName()));
+  }
+}

gateway-server/src/main/java/org/apache/knox/gateway/services/topology/impl/GatewayStatusService.java

@@ -0,0 +1,98 @@
+/*
+ * 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.knox.gateway.services.topology.impl;
+
+import java.io.File;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.commons.io.FilenameUtils;
+import org.apache.knox.gateway.GatewayMessages;
+import org.apache.knox.gateway.config.GatewayConfig;
+import org.apache.knox.gateway.i18n.messages.MessagesFactory;
+import org.apache.knox.gateway.services.Service;
+import org.apache.knox.gateway.services.ServiceLifecycleException;
+
+public class GatewayStatusService implements Service {
+  private static final GatewayMessages LOG = MessagesFactory.get(GatewayMessages.class);
+  private final Set<String> deployedTopologies = new HashSet<>();
+  private Set<String> topologyNamesToCheck = new HashSet<>();
+  private GatewayConfig config;
+
+  public synchronized void onTopologyReady(String topologyName) {
+    deployedTopologies.add(topologyName);
+  }
+
+  public synchronized boolean status() {
+    if (topologyNamesToCheck.isEmpty()) {
+      LOG.noTopologiesToCheck();
+      return false;
+    }
+    Set<String> missing = new HashSet<>(topologyNamesToCheck);
+    missing.removeAll(deployedTopologies);
+    LOG.checkingGatewayStatus(deployedTopologies, missing);
+    return missing.isEmpty();
+  }
+
+  /**
+   * The list of topologies (which will be used to check the gateway status.)
+   * are either coming from the config or collected automatically.
+   * In the later case this should be called at startup, after the hadoop xml resource parser
+   * already generated the descriptors from the hxr
+   */
+  public synchronized void initTopologiesToCheck() {
+    Set<String> healthCheckTopologies = config.getHealthCheckTopologies();
+    if (healthCheckTopologies.isEmpty()) {
+      topologyNamesToCheck = collectTopologies(config);
+    } else {
+      topologyNamesToCheck = healthCheckTopologies;
+    }
+    LOG.startingStatusMonitor(topologyNamesToCheck);
+  }
+
+  private Set<String> collectTopologies(GatewayConfig config) {
+    Set<String> result = new HashSet<>();
+    collectFiles(result, config.getGatewayTopologyDir(), ".xml");
+    collectFiles(result, config.getGatewayDescriptorsDir(), ".json");
+    LOG.collectedTopologiesForHealthCheck(result);
+    return result;
+  }
+
+  private void collectFiles(Set<String> result, String directory, String extension) {
+    File[] files = new File(directory).getAbsoluteFile()
+            .listFiles((dir, name) -> name.toLowerCase(Locale.ROOT).endsWith(extension));
+    if (files != null) {
+      for (File file : files) {
+        result.add(FilenameUtils.getBaseName(file.getName()));
+      }
+    }
+  }
+  @Override
+  public void init(GatewayConfig config, Map<String, String> options) throws ServiceLifecycleException {
+    this.config = config;
+    // this is soon to collect the topologies, topologies are collected in initTopologiesToCheck
+  }
+
+  @Override
+  public void start() throws ServiceLifecycleException {}
+
+  @Override
+  public void stop() throws ServiceLifecycleException {}
+}

gateway-server/src/main/resources/META-INF/services/org.apache.knox.gateway.services.ServiceFactory

@@ -31,4 +31,5 @@ org.apache.knox.gateway.services.factory.SslServiceFactory
 org.apache.knox.gateway.services.factory.TokenServiceFactory
 org.apache.knox.gateway.services.factory.TokenStateServiceFactory
 org.apache.knox.gateway.services.factory.TopologyServiceFactory
-org.apache.knox.gateway.services.factory.ConcurrentSessionVerifierFactory
+org.apache.knox.gateway.services.factory.ConcurrentSessionVerifierFactory
+org.apache.knox.gateway.services.factory.GatewayStatusServiceFactory

gateway-server/src/test/java/org/apache/knox/gateway/services/AbstractGatewayServicesTest.java

@@ -65,6 +65,7 @@ public void testAddStartAndStop() throws ServiceLifecycleException {
         ServiceType.SERVICE_REGISTRY_SERVICE,
         ServiceType.CONCURRENT_SESSION_VERIFIER,
         ServiceType.REMOTE_CONFIGURATION_MONITOR,
+        ServiceType.GATEWAY_STATUS_SERVICE
     };
 
     assertNotEquals(ServiceType.values(), orderedServiceTypes);

gateway-server/src/test/java/org/apache/knox/gateway/services/topology/impl/GatewayStatusServiceTest.java

@@ -0,0 +1,46 @@
+/*
+ * 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.knox.gateway.services.topology.impl;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+import java.util.HashSet;
+
+import org.apache.knox.gateway.config.GatewayConfig;
+import org.easymock.EasyMock;
+import org.junit.Test;
+
+public class GatewayStatusServiceTest {
+
+  @Test
+  public void testReadyStatus() throws Exception {
+    GatewayStatusService statusService = new GatewayStatusService();
+    GatewayConfig config = EasyMock.createNiceMock(GatewayConfig.class);
+    statusService.init(config, null);
+    assertFalse(statusService.status());
+    EasyMock.expect(config.getHealthCheckTopologies()).andReturn(new HashSet<>(Arrays.asList("t1", "t2"))).anyTimes();
+    EasyMock.replay(config);
+    statusService.initTopologiesToCheck();
+    statusService.onTopologyReady("t1");
+    assertFalse(statusService.status());
+    statusService.onTopologyReady("t2");
+    assertTrue(statusService.status());
+  }
+}

gateway-service-health/src/main/java/org/apache/knox/gateway/service/health/PingResource.java

@@ -18,6 +18,9 @@
 package org.apache.knox.gateway.service.health;
 
 import org.apache.knox.gateway.i18n.messages.MessagesFactory;
+import org.apache.knox.gateway.services.GatewayServices;
+import org.apache.knox.gateway.services.ServiceType;
+import org.apache.knox.gateway.services.topology.impl.GatewayStatusService;
 
 import javax.servlet.ServletContext;
 import javax.servlet.http.HttpServletRequest;
@@ -39,8 +42,9 @@
 @Path(PingResource.RESOURCE_PATH)
 public class PingResource {
   static final String VERSION_TAG = "v1";
-  static final String RESOURCE_PATH = "/" + VERSION_TAG + "/ping";
-  public static final String CONTENT = "OK";
+  static final String RESOURCE_PATH = "/" + VERSION_TAG;
+  public static final String OK = "OK";
+  public static final String PENDING = "PENDING";
   private static HealthServiceMessages log = MessagesFactory.get(HealthServiceMessages.class);
   private static final String CONTENT_TYPE = "text/plain";
   private static final String CACHE_CONTROL = "Cache-Control";
@@ -57,12 +61,14 @@ public class PingResource {
 
   @GET
   @Produces({APPLICATION_JSON, TEXT_PLAIN})
+  @Path("ping")
   public Response doGet() {
     return getPingResponse();
   }
 
   @POST
   @Produces({APPLICATION_JSON, TEXT_PLAIN})
+  @Path("ping")
   public Response doPost() {
     return getPingResponse();
   }
@@ -81,7 +87,26 @@ private Response getPingResponse() {
   }
 
   String getPingContent() {
-    return CONTENT;
+    return OK;
+  }
+
+  @GET
+  @Produces({TEXT_PLAIN})
+  @Path("gateway-status")
+  public Response status() {
+    response.setStatus(HttpServletResponse.SC_OK);
+    response.setHeader(CACHE_CONTROL, NO_CACHE);
+    response.setContentType(CONTENT_TYPE);
+    GatewayServices services = (GatewayServices) request.getServletContext()
+            .getAttribute(GatewayServices.GATEWAY_SERVICES_ATTRIBUTE);
+    GatewayStatusService statusService = services.getService(ServiceType.GATEWAY_STATUS_SERVICE);
+    try (PrintWriter writer = response.getWriter()) {
+      writer.println(statusService.status() ? OK : PENDING);
+    } catch (IOException e) {
+      log.logException("status", e);
+      return Response.serverError().entity(String.format(Locale.ROOT, "Failed to reply correctly due to : %s ", e)).build();
+    }
+    return Response.ok().build();
   }
 
 }

gateway-service-health/src/test/java/org/apache/knox/gateway/service/health/PingResourceTest.java

@@ -24,6 +24,6 @@ public class PingResourceTest {
   @Test
   public void testPingCdoGetontent() {
     PingResource pr = new PingResource();
-    Assert.assertEquals(pr.getPingContent(), pr.CONTENT);
+    Assert.assertEquals(pr.getPingContent(), pr.OK);
   }
 }