[cuebot/pycue/proto/sandbox/docs] Add full event-driven monitoring stack, enhance metrics, dashboards, and documentation (#2086)

**Link the Issue(s) this Pull Request is related to.**
- #2085

**Summarize your change.**

[cuebot/pycue/proto/rust/sandbox/docs] Add event-driven monitoring stack
for OpenCue

Implement event-driven monitoring infrastructure enabling real-time and
historical analysis of render farm activity. Adds a Kafka +
Elasticsearch pipeline for collecting job, layer, frame, host, and proc
lifecycle events, with Prometheus and Grafana integration for live
dashboards and operational visibility.

Proto & Event Model:
- Define monitoring.proto with job/layer/frame/host/proc lifecycle
events
- Use proto composition pattern - embed Job, Layer, Frame, Host messages
- Exclude HostReportEvent from pipeline (too high frequency for
Kafka/ES)

Cuebot Event Publishing:
- Add KafkaEventPublisher for async event publishing to Kafka topics
- Add KafkaAdminClient for topic creation with configurable
partitions/retention
- Add MonitoringEventBuilder as Spring-managed bean for event
construction
- Hook publishing into FrameCompleteHandler, HostReportHandler,
DispatchSupportService, JobManagerSupport, DependManagerService
- Publish pickup time tracking events (FRAME_STARTED, FRAME_DISPATCHED)
- Add isFrameDispatchable() to DependDao for dependency checking

Prometheus Metrics:
- cue_frames_completed_total (with show, shot, state labels)
- cue_jobs_completed_total (with show, shot, state labels)
- cue_job_core_seconds histogram
- cue_layer_max_runtime_seconds histogram
- cue_layer_max_memory_bytes histogram

Rust monitoring-indexer Service:
- Add rust/crates/monitoring-indexer: standalone Kafka-to-Elasticsearch
indexer
- Async Kafka consumer with configurable batch processing
- Elasticsearch bulk indexing with date-based indices and field mappings
- Parallel event processing using rayon for CPU-bound operations
- Index templates for all event types (job, layer, frame, host, proc)
- Graceful handling of UnknownTopicOrPartition during startup

gRPC & PyCue:
- Add MonitoringInterface gRPC service
- Implement pycue monitoring wrapper with historical data API methods

Infrastructure (docker-compose.monitoring-full.yml):
- Zookeeper, Kafka, Kafka UI
- Elasticsearch, Kibana
- Prometheus (with cuebot scrape config)
- Grafana (with provisioned dashboard)
- monitoring-indexer service

Grafana Dashboard:
- Frame completion rates by state (DEAD/red, SUCCEEDED/green,
WAITING/yellow)
- Job completion by show
- Frame runtime and memory distribution
- Job core seconds distribution
- Pickup time metrics (FRAME_STARTED/FRAME_DISPATCHED)
- Layer max runtime/memory panels

Documentation:
- Architecture, concepts, and pipeline explanation
- Deployment and Quick Start guides
- User and Developer guides
- API Reference and tutorials
- Elasticsearch query reference guide

Utilities:
- sandbox/monitor_events.py: Example Kafka consumer
- sandbox/load_test_jobs.py: Test data generator with CLI args

Configuration (opt-in, disabled by default):
- monitoring.kafka.enabled, monitoring.kafka.bootstrap.servers
- monitoring.kafka.topic.partitions, .replication.factor, .retention.ms
- monitoring.elasticsearch.enabled, monitoring.elasticsearch.host

---------

Signed-off-by: Ramon Figueiredo <[email protected]>

Author: ramonfigueiredo

.gitignore

@@ -32,4 +32,5 @@ docs/bin/
 sandbox/kafka-data
 sandbox/zookeeper-data
 sandbox/zookeeper-logs
+sandbox/rqd/shots/
 docs/_data/version.yml

VERSION.in

@@ -1 +1 @@
-1.14
+1.15

cuebot/build.gradle

@@ -26,6 +26,8 @@ repositories {
 def grpcVersion = '1.47.0'
 def protobufVersion = '3.21.2'
 def activemqVersion = '5.12.0'
+def kafkaVersion = '3.4.0'
+def elasticsearchVersion = '8.8.0'
 
 // Spring dependency versions are managed by the io.spring.dependency-management plugin.
 // Appropriate versions will be pulled based on the spring boot version specified in the
@@ -52,6 +54,15 @@ dependencies {
     implementation group: 'io.prometheus', name: 'simpleclient', version: '0.16.0'
     implementation group: 'io.prometheus', name: 'simpleclient_servlet', version: '0.16.0'
 
+    // Kafka for event publishing
+    implementation group: 'org.apache.kafka', name: 'kafka-clients', version: "${kafkaVersion}"
+    implementation group: 'org.springframework.kafka', name: 'spring-kafka', version: '2.9.0'
+
+    // Elasticsearch for historical data storage
+    implementation group: 'co.elastic.clients', name: 'elasticsearch-java', version: "${elasticsearchVersion}"
+    implementation group: 'org.elasticsearch.client', name: 'elasticsearch-rest-client', version: "${elasticsearchVersion}"
+    implementation group: 'jakarta.json', name: 'jakarta.json-api', version: '2.1.1'
+
     protobuf files("../proto/src/")
 
     testImplementation group: 'junit', name: 'junit', version: '4.12'
@@ -67,12 +78,14 @@ dependencies {
 
 compileJava {
     dependsOn generateProto
-    options.compilerArgs << "-Xlint:all" << "-Werror"
+    // Exclude serial warning due to protobuf-generated code warnings
+    options.compilerArgs << "-Xlint:all,-serial" << "-Werror"
 }
 
 compileTestJava {
     dependsOn generateProto
-    options.compilerArgs << "-Xlint:all" << "-Werror"
+    // Exclude serial warning due to protobuf-generated code warnings
+    options.compilerArgs << "-Xlint:all,-serial" << "-Werror"
 }
 
 protobuf {

cuebot/src/main/java/com/imageworks/spcue/CuebotApplication.java

@@ -24,8 +24,9 @@
 import org.apache.logging.log4j.LogManager;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
 
-@SpringBootApplication
+@SpringBootApplication(exclude = {KafkaAutoConfiguration.class})
 public class CuebotApplication extends SpringApplication {
     private static String[] checkArgs(String[] args) {
         Optional<String> deprecatedFlag = Arrays.stream(args)

cuebot/src/main/java/com/imageworks/spcue/ExecutionSummary.java

@@ -28,6 +28,15 @@ public class ExecutionSummary {
     public long gpuTimeSuccess;
     public long gpuTimeFail;
     public long highMemoryKb;
+    public int highFrameSec;
+
+    public int getHighFrameSec() {
+        return highFrameSec;
+    }
+
+    public void setHighFrameSec(int highFrameSec) {
+        this.highFrameSec = highFrameSec;
+    }
 
     public long getHighMemoryKb() {
         return highMemoryKb;

cuebot/src/main/java/com/imageworks/spcue/PrometheusMetricsCollector.java

@@ -119,6 +119,38 @@ public class PrometheusMetricsCollector {
             .labelNames("env", "cuebot_host", "render_node", "job_name", "frame_name", "frame_id")
             .register();
 
+    private static final Counter frameCompletedCounter = Counter.build()
+            .name("cue_frames_completed_total").help("Total number of frames completed")
+            .labelNames("env", "cuebot_host", "state", "show", "shot").register();
+
+    private static final Counter jobCompletedCounter =
+            Counter.build().name("cue_jobs_completed_total").help("Total number of jobs completed")
+                    .labelNames("env", "cuebot_host", "state", "show", "shot").register();
+
+    private static final Histogram jobCoreSecondsHistogram = Histogram.build()
+            .name("cue_job_core_seconds").help("Histogram of total core seconds per job")
+            .labelNames("env", "cuebot_host", "show", "shot")
+            .buckets(3600, 36000, 360000, 3600000, 36000000).register();
+
+    private static final Histogram layerMaxRuntimeHistogram =
+            Histogram.build().name("cue_layer_max_runtime_seconds")
+                    .help("Histogram of max frame runtime per layer in seconds")
+                    .labelNames("env", "cuebot_host", "show", "shot", "layer_type")
+                    .buckets(60, 300, 600, 1800, 3600, 7200, 14400, 28800, 86400).register();
+
+    private static final Histogram layerMaxMemoryHistogram =
+            Histogram.build().name("cue_layer_max_memory_bytes")
+                    .help("Histogram of max frame memory usage per layer in bytes")
+                    .labelNames("env", "cuebot_host", "show", "shot", "layer_type")
+                    .buckets(256L * 1024 * 1024, 512L * 1024 * 1024, 1024L * 1024 * 1024,
+                            2048L * 1024 * 1024, 4096L * 1024 * 1024, 8192L * 1024 * 1024,
+                            16384L * 1024 * 1024, 32768L * 1024 * 1024)
+                    .register();
+
+    private static final Counter hostReportsReceivedCounter = Counter.build()
+            .name("cue_host_reports_received_total").help("Total number of host reports received")
+            .labelNames("env", "cuebot_host", "facility").register();
+
     private String deployment_environment;
     private String cuebot_host;
 
@@ -269,6 +301,82 @@ public void incrementFrameKillFailureCounter(String hostname, String jobName, St
                 jobName, frameName, frameId).inc();
     }
 
+    /**
+     * Record a frame completion
+     *
+     * @param state final state of the frame
+     * @param show show name
+     * @param shot shot name
+     */
+    public void recordFrameCompleted(String state, String show, String shot) {
+        frameCompletedCounter
+                .labels(this.deployment_environment, this.cuebot_host, state, show, shot).inc();
+    }
+
+    /**
+     * Record a job completion
+     *
+     * @param state final state of the job
+     * @param show show name
+     * @param shot shot name
+     */
+    public void recordJobCompleted(String state, String show, String shot) {
+        jobCompletedCounter.labels(this.deployment_environment, this.cuebot_host, state, show, shot)
+                .inc();
+    }
+
+    /**
+     * Record job total core seconds for histogramming
+     *
+     * @param coreSeconds total core seconds consumed by the job
+     * @param show show name
+     * @param shot shot name
+     */
+    public void recordJobCoreSeconds(double coreSeconds, String show, String shot) {
+        jobCoreSecondsHistogram.labels(this.deployment_environment, this.cuebot_host, show, shot)
+                .observe(coreSeconds);
+    }
+
+    /**
+     * Record layer max runtime for histogramming
+     *
+     * @param runtimeSeconds max runtime in seconds for the layer
+     * @param show show name
+     * @param shot shot name
+     * @param layerType layer type
+     */
+    public void recordLayerMaxRuntime(double runtimeSeconds, String show, String shot,
+            String layerType) {
+        layerMaxRuntimeHistogram
+                .labels(this.deployment_environment, this.cuebot_host, show, shot, layerType)
+                .observe(runtimeSeconds);
+    }
+
+    /**
+     * Record layer max memory usage for histogramming
+     *
+     * @param memoryBytes max memory in bytes for the layer
+     * @param show show name
+     * @param shot shot name
+     * @param layerType layer type
+     */
+    public void recordLayerMaxMemory(double memoryBytes, String show, String shot,
+            String layerType) {
+        layerMaxMemoryHistogram
+                .labels(this.deployment_environment, this.cuebot_host, show, shot, layerType)
+                .observe(memoryBytes);
+    }
+
+    /**
+     * Record a host report received
+     *
+     * @param facility facility name
+     */
+    public void recordHostReport(String facility) {
+        hostReportsReceivedCounter.labels(this.deployment_environment, this.cuebot_host, facility)
+                .inc();
+    }
+
     // Setters used for dependency injection
     public void setBookingQueue(BookingQueue bookingQueue) {
         this.bookingQueue = bookingQueue;

cuebot/src/main/java/com/imageworks/spcue/config/AppConfig.java

@@ -38,7 +38,8 @@
         "classpath:conf/spring/applicationContext-grpcServer.xml",
         "classpath:conf/spring/applicationContext-service.xml",
         "classpath:conf/spring/applicationContext-jms.xml",
-        "classpath:conf/spring/applicationContext-criteria.xml"})
+        "classpath:conf/spring/applicationContext-criteria.xml",
+        "classpath:conf/spring/applicationContext-monitoring.xml"})
 @EnableConfigurationProperties
 @PropertySource({"classpath:opencue.properties"})
 public class AppConfig {

cuebot/src/main/java/com/imageworks/spcue/dao/DependDao.java

@@ -194,6 +194,14 @@ public interface DependDao {
      */
     boolean decrementDependCount(FrameInterface f);
 
+    /**
+     * Check if a frame is dispatchable (has depend_count = 0).
+     *
+     * @param f the frame to check
+     * @return true if the frame's depend_count is 0
+     */
+    boolean isFrameDispatchable(FrameInterface f);
+
     /**
      * Returns true if this is the thread that set the depend to inactive.
      *

cuebot/src/main/java/com/imageworks/spcue/dao/postgres/DependDaoJdbc.java

@@ -215,6 +215,15 @@ public boolean decrementDependCount(FrameInterface f) {
         return getJdbcTemplate().update(DECREMENT_DEPEND_COUNT, f.getFrameId()) == 1;
     }
 
+    private static final String IS_FRAME_DISPATCHABLE =
+            "SELECT int_depend_count = 0 FROM frame WHERE pk_frame = ?";
+
+    @Override
+    public boolean isFrameDispatchable(FrameInterface f) {
+        return Boolean.TRUE.equals(getJdbcTemplate().queryForObject(IS_FRAME_DISPATCHABLE,
+                Boolean.class, f.getFrameId()));
+    }
+
     private static final String[] DELETE_DEPEND =
             {"DELETE FROM depend WHERE pk_parent=?", "DELETE FROM depend WHERE pk_depend=?"};
 

cuebot/src/main/java/com/imageworks/spcue/dao/postgres/LayerDaoJdbc.java

@@ -409,12 +409,13 @@ public FrameStateTotals mapRow(ResultSet rs, int rowNum) throws SQLException {
                 }, layer.getLayerId());
     }
 
-    private static final String GET_EXECUTION_SUMMARY = "SELECT "
-            + "layer_usage.int_core_time_success," + "layer_usage.int_core_time_fail,"
-            + "layer_usage.int_gpu_time_success," + "layer_usage.int_gpu_time_fail,"
-            + "layer_usage.int_clock_time_success," + "layer_mem.int_max_rss " + "FROM " + "layer,"
-            + "layer_usage, " + "layer_mem " + "WHERE " + "layer.pk_layer = layer_usage.pk_layer "
-            + "AND " + "layer.pk_layer = layer_mem.pk_layer " + "AND " + "layer.pk_layer = ?";
+    private static final String GET_EXECUTION_SUMMARY =
+            "SELECT " + "layer_usage.int_core_time_success," + "layer_usage.int_core_time_fail,"
+                    + "layer_usage.int_gpu_time_success," + "layer_usage.int_gpu_time_fail,"
+                    + "layer_usage.int_clock_time_success," + "layer_usage.int_clock_time_high,"
+                    + "layer_mem.int_max_rss " + "FROM " + "layer," + "layer_usage, " + "layer_mem "
+                    + "WHERE " + "layer.pk_layer = layer_usage.pk_layer " + "AND "
+                    + "layer.pk_layer = layer_mem.pk_layer " + "AND " + "layer.pk_layer = ?";
 
     @Override
     public ExecutionSummary getExecutionSummary(LayerInterface layer) {
@@ -429,6 +430,7 @@ public ExecutionSummary mapRow(ResultSet rs, int rowNum) throws SQLException {
                         e.gpuTimeFail = rs.getLong("int_gpu_time_fail");
                         e.gpuTime = e.gpuTimeSuccess + e.gpuTimeFail;
                         e.highMemoryKb = rs.getLong("int_max_rss");
+                        e.highFrameSec = rs.getInt("int_clock_time_high");
                         return e;
                     }
                 }, layer.getLayerId());