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

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -903,7 +903,7 @@ class BeamModulePlugin implements Plugin<Project> {
         testcontainers_rabbitmq                     : "org.testcontainers:rabbitmq:$testcontainers_version",
         truth                                       : "com.google.truth:truth:1.1.5",
         threetenbp                                  : "org.threeten:threetenbp:1.6.8",
-        vendored_grpc_1_60_1                        : "org.apache.beam:beam-vendor-grpc-1_60_1:0.1",
+        vendored_grpc_1_60_1                        : "org.apache.beam:beam-vendor-grpc-1_60_1:0.2",
         vendored_guava_32_1_2_jre                   : "org.apache.beam:beam-vendor-guava-32_1_2-jre:0.1",
         vendored_calcite_1_28_0                     : "org.apache.beam:beam-vendor-calcite-1_28_0:0.2",
         woodstox_core_asl                           : "org.codehaus.woodstox:woodstox-core-asl:4.4.1",

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

@@ -146,7 +146,7 @@ protected BatchDataflowWorker(
 
     this.memoryMonitor = MemoryMonitor.fromOptions(options);
     this.statusPages =
-        WorkerStatusPages.create(DEFAULT_STATUS_PORT, this.memoryMonitor, () -> true);
+        WorkerStatusPages.create(options, DEFAULT_STATUS_PORT, this.memoryMonitor, () -> true);
 
     if (!DataflowRunner.hasExperiment(options, "disable_debug_capture")) {
       this.debugCaptureManager =

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

@@ -314,7 +314,8 @@ public class StreamingDataflowWorker {
     this.executorSupplier = executorSupplier;
     this.windmillServiceEnabled = options.isEnableStreamingEngine();
     this.memoryMonitor = MemoryMonitor.fromOptions(options);
-    this.statusPages = WorkerStatusPages.create(DEFAULT_STATUS_PORT, memoryMonitor, () -> true);
+    this.statusPages =
+        WorkerStatusPages.create(options, DEFAULT_STATUS_PORT, memoryMonitor, () -> true);
     if (windmillServiceEnabled) {
       this.debugCaptureManager =
           new DebugCapture.Manager(options, statusPages.getDebugCapturePages());

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

@@ -130,6 +130,12 @@ public interface StreamingDataflowWorkerOptions extends DataflowWorkerHarnessOpt
 
   void setWindmillMessagesBetweenIsReadyChecks(int value);
 
+  @Description("If true, will only show windmill service channels on /channelz")
+  @Default.Boolean(true)
+  boolean getChannelzShowOnlyWindmillServiceChannels();
+
+  void setChannelzShowOnlyWindmillServiceChannels(boolean value);
+
   /**
    * Factory for creating local Windmill address. Reads from system propery 'windmill.hostport' for
    * backwards compatibility.

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

@@ -26,8 +26,11 @@
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import org.apache.beam.runners.core.construction.Environments;
+import org.apache.beam.runners.dataflow.options.DataflowWorkerHarnessOptions;
+import org.apache.beam.runners.dataflow.worker.options.StreamingDataflowWorkerOptions;
 import org.apache.beam.runners.dataflow.worker.status.DebugCapture.Capturable;
 import org.apache.beam.runners.dataflow.worker.util.MemoryMonitor;
+import org.apache.beam.runners.dataflow.worker.windmill.client.grpc.ChannelzServlet;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.servlet.ServletHandler;
@@ -50,13 +53,18 @@ public class WorkerStatusPages {
   private final ServletHandler servletHandler = new ServletHandler();
 
   @VisibleForTesting
-  WorkerStatusPages(Server server, MemoryMonitor memoryMonitor, BooleanSupplier healthyIndicator) {
+  WorkerStatusPages(
+      DataflowWorkerHarnessOptions options,
+      Server server,
+      MemoryMonitor memoryMonitor,
+      BooleanSupplier healthyIndicator) {
     this.statusServer = server;
     this.capturePages = new ArrayList<>();
     this.statusServer.setHandler(servletHandler);
 
     // Install the default servlets (threadz, healthz, heapz, jfrz, statusz)
     addServlet(threadzServlet);
+
     addServlet(new HealthzServlet(healthyIndicator));
     addServlet(new HeapzServlet(memoryMonitor));
     if (Environments.getJavaVersion() != Environments.JavaVersion.java8) {
@@ -67,17 +75,26 @@ public class WorkerStatusPages {
     // Add default capture pages (threadz, statusz)
     this.capturePages.add(threadzServlet);
     this.capturePages.add(statuszServlet);
+    if (options.isStreaming()) {
+      ChannelzServlet channelzServlet =
+          new ChannelzServlet(options.as(StreamingDataflowWorkerOptions.class));
+      addServlet(channelzServlet);
+      this.capturePages.add(channelzServlet);
+    }
     // Add some status pages
     addStatusDataProvider("resources", "Resources", memoryMonitor);
   }
 
   public static WorkerStatusPages create(
-      int defaultStatusPort, MemoryMonitor memoryMonitor, BooleanSupplier healthyIndicator) {
+      DataflowWorkerHarnessOptions options,
+      int defaultStatusPort,
+      MemoryMonitor memoryMonitor,
+      BooleanSupplier healthyIndicator) {
     int statusPort = defaultStatusPort;
     if (System.getProperties().containsKey("status_port")) {
       statusPort = Integer.parseInt(System.getProperty("status_port"));
     }
-    return new WorkerStatusPages(new Server(statusPort), memoryMonitor, healthyIndicator);
+    return new WorkerStatusPages(options, new Server(statusPort), memoryMonitor, healthyIndicator);
   }
 
   /** Start the server. */

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

@@ -23,6 +23,7 @@
 import org.apache.beam.runners.dataflow.worker.windmill.client.WindmillStream.GetDataStream;
 import org.apache.beam.runners.dataflow.worker.windmill.client.WindmillStream.GetWorkStream;
 import org.apache.beam.runners.dataflow.worker.windmill.work.WorkItemReceiver;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.net.HostAndPort;
 
 /**
@@ -53,6 +54,11 @@ public void setWindmillServiceEndpoints(Set<HostAndPort> endpoints) throws IOExc
     // This class is used for windmill appliance and local runner tests.
   }
 
+  @Override
+  public ImmutableSet<HostAndPort> getWindmillServiceEndpoints() {
+    return ImmutableSet.of();
+  }
+
   @Override
   public boolean isReady() {
     return true;

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

@@ -28,6 +28,7 @@
 import org.apache.beam.runners.dataflow.worker.windmill.client.WindmillStream.GetDataStream;
 import org.apache.beam.runners.dataflow.worker.windmill.client.WindmillStream.GetWorkStream;
 import org.apache.beam.runners.dataflow.worker.windmill.work.WorkItemReceiver;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.net.HostAndPort;
 
 /** Stub for communicating with a Windmill server. */
@@ -43,6 +44,11 @@ public abstract class WindmillServerStub implements StatusDataProvider {
    */
   public abstract void setWindmillServiceEndpoints(Set<HostAndPort> endpoints) throws IOException;
 
+  /*
+   * Returns the windmill service endpoints set by setWindmillServiceEndpoints
+   */
+  public abstract ImmutableSet<HostAndPort> getWindmillServiceEndpoints();
+
   /** Returns true iff this WindmillServerStub is ready for making API calls. */
   public abstract boolean isReady();