[Dataflow Streaming] Add Channelz staus page exporting GRPC channelz data

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

@@ -212,6 +212,7 @@ public class StreamingDataflowWorker {
 
   private static final Duration MAX_LOCAL_PROCESSING_RETRY_DURATION = Duration.standardMinutes(5);
   private static final Random clientIdGenerator = new Random();
+  private static final String CHANNELZ_PATH = "/channelz";
   final WindmillStateCache stateCache;
   // Maps from computation ids to per-computation state.
   private final ConcurrentMap<String, ComputationState> computationMap;
@@ -738,7 +739,7 @@ public void startStatusPages() {
     }
 
     if (windmillServiceEnabled) {
-      ChannelzServlet channelzServlet = new ChannelzServlet(options, windmillServer);
+      ChannelzServlet channelzServlet = new ChannelzServlet(CHANNELZ_PATH, options, windmillServer);
       statusPages.addServlet(channelzServlet);
       statusPages.addCapturePage(channelzServlet);
     }
@@ -2089,6 +2090,7 @@ public void doGet(HttpServletRequest request, HttpServletResponse response) thro
   }
 
   private class MetricsDataProvider implements StatusDataProvider {
+
     @Override
     public void appendSummaryHtml(PrintWriter writer) {
       writer.println(workUnitExecutor.summaryHtml());

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/ChannelzServlet.java

@@ -39,18 +39,18 @@
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.net.HostAndPort;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.SettableFuture;
 
-/** Respond to /channelz with the GRPC channelz data. */
+/** Respond to /path with the GRPC channelz data. */
 public class ChannelzServlet extends BaseStatusServlet implements DebugCapture.Capturable {
-  private static final String PATH = "/channelz";
-  private static final int MAX_TOP_CHANNELS_TO_RETURN = 100;
+
+  private static final int MAX_TOP_CHANNELS_TO_RETURN = 500;
 
   private final ChannelzService channelzService;
   private final WindmillServerStub windmillServerStub;
   private final boolean showOnlyWindmillServiceChannels;
 
   public ChannelzServlet(
-      StreamingDataflowWorkerOptions options, WindmillServerStub windmillServerStub) {
-    super(PATH);
+      String path, StreamingDataflowWorkerOptions options, WindmillServerStub windmillServerStub) {
+    super(path);
     channelzService = ChannelzService.newInstance(MAX_TOP_CHANNELS_TO_RETURN);
     this.windmillServerStub = windmillServerStub;
     showOnlyWindmillServiceChannels = options.getChannelzShowOnlyWindmillServiceChannels();
@@ -66,7 +66,7 @@ protected void doGet(HttpServletRequest request, HttpServletResponse response)
 
   @Override
   public String pageName() {
-    return PATH;
+    return getPath();
   }
 
   @Override
@@ -80,12 +80,17 @@ public void captureData(PrintWriter writer) {
   // channelz proto says there won't be cycles in the ref graph.
   // we track visited ids to be defensive and prevent any accidental cycles.
   static class VisitedSets {
+
     Set<Long> channels = new HashSet<>();
     Set<Long> subchannels = new HashSet<>();
   }
 
   private void appendTopChannels(PrintWriter writer) {
     SettableFuture<GetTopChannelsResponse> future = SettableFuture.create();
+    // IDEA: If there are more than MAX_TOP_CHANNELS_TO_RETURN top channels
+    // in the worker, we might not return all the windmill channels. If we run into
+    // such situations, this logic can be modified to loop till we see an empty
+    // GetTopChannelsResponse response with the end bit set.
     channelzService.getTopChannels(
         GetTopChannelsRequest.newBuilder().build(), getStreamObserver(future));
     GetTopChannelsResponse topChannelsResponse;

runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/ChannelzServletTest.java

@@ -56,7 +56,7 @@ public void testRendersAllChannels() throws UnsupportedEncodingException {
     fakeWindmillServer.setWindmillServiceEndpoints(
         ImmutableSet.of(HostAndPort.fromHost(windmill1), HostAndPort.fromHost(windmill2)));
     options.setChannelzShowOnlyWindmillServiceChannels(false);
-    ChannelzServlet channelzServlet = new ChannelzServlet(options, fakeWindmillServer);
+    ChannelzServlet channelzServlet = new ChannelzServlet("/channelz", options, fakeWindmillServer);
     StringWriter stringWriter = new StringWriter();
     PrintWriter writer = new PrintWriter(stringWriter);
     channelzServlet.captureData(writer);
@@ -88,7 +88,7 @@ public void testRendersOnlyWindmillChannels() throws UnsupportedEncodingExceptio
     fakeWindmillServer.setWindmillServiceEndpoints(
         ImmutableSet.of(HostAndPort.fromHost(windmill1), HostAndPort.fromHost(windmill2)));
     options.setChannelzShowOnlyWindmillServiceChannels(true);
-    ChannelzServlet channelzServlet = new ChannelzServlet(options, fakeWindmillServer);
+    ChannelzServlet channelzServlet = new ChannelzServlet("/channelz", options, fakeWindmillServer);
     StringWriter stringWriter = new StringWriter();
     PrintWriter writer = new PrintWriter(stringWriter);
     channelzServlet.captureData(writer);