add light weight version of /druid/coordinator/v1/lookups/nodeStatus (#10422)

* add light weight version /druid/coordinator/v1/lookups/nodeStatus

* review stuffs

docs/querying/lookups.md

@@ -347,7 +347,7 @@ These end points can be used to get the propagation status of configured lookups
 
 ### List lookup state of all processes
 
-`GET /druid/coordinator/v1/lookups/nodeStatus` with optional query parameter `discover` to discover tiers or configured lookup tiers are listed.
+`GET /druid/coordinator/v1/lookups/nodeStatus` with optional query parameter `discover` to discover tiers advertised by other Druid nodes, or by default, returning all configured lookup tiers. The default response will also include the lookups which are loaded, being loaded, or being dropped on each node, for each tier, including the complete lookup spec. Add the optional query parameter `detailed=false` to only include the 'version' of the lookup instead of the complete spec.
 
 ### List lookup state of processes in a tier
 

server/src/main/java/org/apache/druid/server/http/LookupCoordinatorResource.java

@@ -42,6 +42,7 @@ import org.apache.druid.server.http.security.ConfigResourceFilter;
 import org.apache.druid.server.lookup.cache.LookupCoordinatorManager;
 import org.apache.druid.server.lookup.cache.LookupExtractorFactoryMapContainer;
 
+import javax.annotation.Nullable;
 import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.Consumes;
 import javax.ws.rs.DELETE;
@@ -66,6 +67,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  * Contains information about lookups exposed through the coordinator
@@ -540,9 +542,12 @@ public class LookupCoordinatorResource
   @Produces({MediaType.APPLICATION_JSON})
   @Path("/nodeStatus")
   public Response getAllNodesStatus(
-      @QueryParam("discover") boolean discover
+      @QueryParam("discover") boolean discover,
+      @QueryParam("detailed") @Nullable Boolean detailed
   )
   {
+    boolean full = detailed == null || detailed;
+
     try {
       Collection<String> tiers;
       if (discover) {
@@ -558,27 +563,16 @@ public class LookupCoordinatorResource
         tiers = configuredLookups.keySet();
       }
 
-      Map<HostAndPort, LookupsState<LookupExtractorFactoryMapContainer>> lookupsStateOnHosts = lookupCoordinatorManager
-          .getLastKnownLookupsStateOnNodes();
+      Map<HostAndPort, LookupsState<LookupExtractorFactoryMapContainer>> lookupsStateOnHosts =
+          lookupCoordinatorManager.getLastKnownLookupsStateOnNodes();
 
-      Map<String, Map<HostAndPort, LookupsState<LookupExtractorFactoryMapContainer>>> result = new HashMap<>();
-
-      for (String tier : tiers) {
-        Map<HostAndPort, LookupsState<LookupExtractorFactoryMapContainer>> tierNodesStatus = new HashMap<>();
-        result.put(tier, tierNodesStatus);
-
-        Collection<HostAndPort> nodes = lookupCoordinatorManager.discoverNodesInTier(tier);
-
-        for (HostAndPort node : nodes) {
-          LookupsState<LookupExtractorFactoryMapContainer> lookupsState = lookupsStateOnHosts.get(node);
-          if (lookupsState == null) {
-            tierNodesStatus.put(node, new LookupsState<>(null, null, null));
-          } else {
-            tierNodesStatus.put(node, lookupsState);
-          }
-        }
+      final Map result;
+      if (full) {
+        result = getDetailedAllNodeStatus(lookupsStateOnHosts, tiers);
+      } else {
+        // lookups to load per host by version
+        result = getSimpleAllNodeStatus(lookupsStateOnHosts, tiers);
       }
-
       return Response.ok(result).build();
     }
     catch (Exception ex) {
@@ -647,6 +641,82 @@ public class LookupCoordinatorResource
     }
   }
 
+
+
+  /**
+   * Build 'simple' lookup cluster status, broken down by tier, host, and then the {@link LookupsState} with
+   * the lookup name and version ({@link LookupExtractorFactoryMapContainer#version})
+   */
+  private Map<String, Map<HostAndPort, LookupsState<String>>> getSimpleAllNodeStatus(
+      Map<HostAndPort, LookupsState<LookupExtractorFactoryMapContainer>> lookupsStateOnHosts,
+      Collection<String> tiers
+  )
+  {
+    Map<String, Map<HostAndPort, LookupsState<String>>> results = new HashMap<>();
+    for (String tier : tiers) {
+      Map<HostAndPort, LookupsState<String>> tierNodesStatus = new HashMap<>();
+      results.put(tier, tierNodesStatus);
+
+      Collection<HostAndPort> nodes = lookupCoordinatorManager.discoverNodesInTier(tier);
+
+      for (HostAndPort node : nodes) {
+        LookupsState<LookupExtractorFactoryMapContainer> lookupsState = lookupsStateOnHosts.get(node);
+        if (lookupsState == null) {
+          tierNodesStatus.put(node, new LookupsState<>(null, null, null));
+        } else {
+          Map<String, String> current = lookupsState.getCurrent()
+                                                    .entrySet()
+                                                    .stream()
+                                                    .collect(
+                                                        Collectors.toMap(
+                                                            Map.Entry::getKey,
+                                                            e -> e.getValue().getVersion()
+                                                        )
+                                                    );
+          Map<String, String> toLoad = lookupsState.getToLoad()
+                                                   .entrySet()
+                                                   .stream()
+                                                   .collect(
+                                                       Collectors.toMap(
+                                                           Map.Entry::getKey,
+                                                           e -> e.getValue().getVersion()
+                                                       )
+                                                   );
+          tierNodesStatus.put(node, new LookupsState<>(current, toLoad, lookupsState.getToDrop()));
+        }
+      }
+    }
+    return results;
+  }
+
+  /**
+   * Build 'detailed' lookup cluster status, broken down by tier, host, and then the full {@link LookupsState} with
+   * complete {@link LookupExtractorFactoryMapContainer} information.
+   */
+  private Map<String, Map<HostAndPort, LookupsState<LookupExtractorFactoryMapContainer>>> getDetailedAllNodeStatus(
+      Map<HostAndPort, LookupsState<LookupExtractorFactoryMapContainer>> lookupsStateOnHosts,
+      Collection<String> tiers
+  )
+  {
+    Map<String, Map<HostAndPort, LookupsState<LookupExtractorFactoryMapContainer>>> result = new HashMap<>();
+    for (String tier : tiers) {
+      Map<HostAndPort, LookupsState<LookupExtractorFactoryMapContainer>> tierNodesStatus = new HashMap<>();
+      result.put(tier, tierNodesStatus);
+
+      Collection<HostAndPort> nodes = lookupCoordinatorManager.discoverNodesInTier(tier);
+
+      for (HostAndPort node : nodes) {
+        LookupsState<LookupExtractorFactoryMapContainer> lookupsState = lookupsStateOnHosts.get(node);
+        if (lookupsState == null) {
+          tierNodesStatus.put(node, new LookupsState<>(null, null, null));
+        } else {
+          tierNodesStatus.put(node, lookupsState);
+        }
+      }
+    }
+    return result;
+  }
+
   @VisibleForTesting
   static class LookupStatus
   {

server/src/test/java/org/apache/druid/server/http/LookupCoordinatorResourceTest.java

@@ -1058,7 +1058,7 @@ public class LookupCoordinatorResourceTest
         MAPPER
     );
 
-    final Response response = lookupCoordinatorResource.getAllNodesStatus(false);
+    final Response response = lookupCoordinatorResource.getAllNodesStatus(false, null);
     Assert.assertEquals(200, response.getStatus());
     Assert.assertEquals(
         ImmutableMap.of(
@@ -1067,7 +1067,44 @@ public class LookupCoordinatorResourceTest
                 LOOKUP_NODE,
                 LOOKUP_STATE
             )
-        ), response.getEntity()
+        ),
+        response.getEntity()
+    );
+
+    EasyMock.verify(lookupCoordinatorManager);
+  }
+
+  @Test
+  public void testGetAllNodesStatusDetailedFalse()
+  {
+    final LookupCoordinatorManager lookupCoordinatorManager = EasyMock.createStrictMock(
+        LookupCoordinatorManager.class
+    );
+    EasyMock.expect(lookupCoordinatorManager.getKnownLookups()).andReturn(SINGLE_TIER_MAP);
+    EasyMock.expect(lookupCoordinatorManager.getLastKnownLookupsStateOnNodes()).andReturn(NODES_LOOKUP_STATE);
+    EasyMock.expect(lookupCoordinatorManager.discoverNodesInTier(LOOKUP_TIER)).andReturn(ImmutableList.of(LOOKUP_NODE));
+
+    EasyMock.replay(lookupCoordinatorManager);
+
+    final LookupCoordinatorResource lookupCoordinatorResource = new LookupCoordinatorResource(
+        lookupCoordinatorManager,
+        MAPPER,
+        MAPPER
+    );
+
+    final Response response = lookupCoordinatorResource.getAllNodesStatus(false, false);
+    Assert.assertEquals(200, response.getStatus());
+    Assert.assertEquals(
+        ImmutableMap.of(
+            LOOKUP_TIER,
+            ImmutableMap.of(
+                LOOKUP_NODE,
+                new LookupsState(
+                    ImmutableMap.of(LOOKUP_NAME, SINGLE_LOOKUP.getVersion()), null, null
+                )
+            )
+        ),
+        response.getEntity()
     );
 
     EasyMock.verify(lookupCoordinatorManager);