[cuebot/rust/sandbox] Decouple Kafka consumer into standalone Rust indexer

Move Kafka-to-Elasticsearch event indexing from Cuebot to a standalone Rust service, addressing code review feedback to decouple the consumer from the Java codebase.

Rust kafka-es-indexer:
- Add rust/crates/kafka-es-indexer: standalone Kafka consumer that indexes OpenCue events (job, layer, frame, host, proc) to Elasticsearch
- Async Kafka consumer with configurable batch processing
- Elasticsearch bulk indexing with date-based indices
- Index templates with proper field mappings for all event types
- CLI with environment variable configuration

Cuebot cleanup:
- Remove Java KafkaEventConsumer and ElasticsearchClient classes
- Remove getJobHistory, getFrameHistory, getLayerHistory, getLayerMemoryHistory from HistoricalDao and HistoricalManager
- Update ManageMonitoring gRPC servant to return UNIMPLEMENTED with message directing users to query Elasticsearch directly
- Keep KafkaEventPublisher for publishing events from Cuebot to Kafka
- Keep core job archival methods (getFinishedJobs, transferJob) intact

Infrastructure:
- Update docker-compose.monitoring-full.yml to include kafka-es-indexer

Author: ramonfigueiredo

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

@@ -18,12 +18,6 @@
 import java.util.List;
 
 import com.imageworks.spcue.JobInterface;
-import com.imageworks.spcue.grpc.job.FrameState;
-import com.imageworks.spcue.grpc.job.JobState;
-import com.imageworks.spcue.grpc.monitoring.HistoricalFrame;
-import com.imageworks.spcue.grpc.monitoring.HistoricalJob;
-import com.imageworks.spcue.grpc.monitoring.HistoricalLayer;
-import com.imageworks.spcue.grpc.monitoring.LayerMemoryRecord;
 
 public interface HistoricalDao {
 
@@ -42,29 +36,4 @@ public interface HistoricalDao {
      */
     void transferJob(JobInterface job);
 
-    /**
-     * Query historical job records from the job_history table.
-     */
-    List<HistoricalJob> getJobHistory(List<String> shows, List<String> users, List<String> shots,
-            List<String> jobNameRegex, List<JobState> states, long startTime, long endTime,
-            int page, int pageSize, int maxResults);
-
-    /**
-     * Query historical frame records from the frame_history table.
-     */
-    List<HistoricalFrame> getFrameHistory(String jobId, String jobName, List<String> layerNames,
-            List<FrameState> states, long startTime, long endTime, int page, int pageSize);
-
-    /**
-     * Query historical layer records from the layer_history table.
-     */
-    List<HistoricalLayer> getLayerHistory(String jobId, String jobName, long startTime,
-            long endTime, int page, int pageSize);
-
-    /**
-     * Query historical memory usage for a layer type.
-     */
-    List<LayerMemoryRecord> getLayerMemoryHistory(String layerName, List<String> shows,
-            long startTime, long endTime, int maxResults);
-
 }

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

@@ -15,27 +15,16 @@
 
 package com.imageworks.spcue.dao.postgres;
 
-import java.util.Collections;
 import java.util.List;
 
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.jdbc.core.support.JdbcDaoSupport;
 
 import com.imageworks.spcue.JobInterface;
 import com.imageworks.spcue.dao.HistoricalDao;
-import com.imageworks.spcue.grpc.job.FrameState;
 import com.imageworks.spcue.grpc.job.JobState;
-import com.imageworks.spcue.grpc.monitoring.HistoricalFrame;
-import com.imageworks.spcue.grpc.monitoring.HistoricalJob;
-import com.imageworks.spcue.grpc.monitoring.HistoricalLayer;
-import com.imageworks.spcue.grpc.monitoring.LayerMemoryRecord;
-import com.imageworks.spcue.monitoring.ElasticsearchClient;
 
 public class HistoricalDaoJdbc extends JdbcDaoSupport implements HistoricalDao {
 
-    @Autowired(required = false)
-    private ElasticsearchClient elasticsearchClient;
-
     private static final String GET_FINISHED_JOBS = JobDaoJdbc.GET_JOB + "WHERE "
             + "job.str_state = ? " + "AND " + "current_timestamp - job.ts_stopped > ";
 
@@ -51,46 +40,4 @@ public void transferJob(JobInterface job) {
          */
         getJdbcTemplate().update("DELETE FROM job WHERE pk_job=?", job.getJobId());
     }
-
-    @Override
-    public List<HistoricalJob> getJobHistory(List<String> shows, List<String> users,
-            List<String> shots, List<String> jobNameRegex, List<JobState> states, long startTime,
-            long endTime, int page, int pageSize, int maxResults) {
-        if (elasticsearchClient != null && elasticsearchClient.isEnabled()) {
-            return elasticsearchClient.searchJobHistory(shows, users, shots, jobNameRegex, states,
-                    startTime, endTime, page, pageSize, maxResults);
-        }
-        return Collections.emptyList();
-    }
-
-    @Override
-    public List<HistoricalFrame> getFrameHistory(String jobId, String jobName,
-            List<String> layerNames, List<FrameState> states, long startTime, long endTime,
-            int page, int pageSize) {
-        if (elasticsearchClient != null && elasticsearchClient.isEnabled()) {
-            return elasticsearchClient.searchFrameHistory(jobId, jobName, layerNames, states,
-                    startTime, endTime, page, pageSize);
-        }
-        return Collections.emptyList();
-    }
-
-    @Override
-    public List<HistoricalLayer> getLayerHistory(String jobId, String jobName, long startTime,
-            long endTime, int page, int pageSize) {
-        if (elasticsearchClient != null && elasticsearchClient.isEnabled()) {
-            return elasticsearchClient.searchLayerHistory(jobId, jobName, startTime, endTime, page,
-                    pageSize);
-        }
-        return Collections.emptyList();
-    }
-
-    @Override
-    public List<LayerMemoryRecord> getLayerMemoryHistory(String layerName, List<String> shows,
-            long startTime, long endTime, int maxResults) {
-        if (elasticsearchClient != null && elasticsearchClient.isEnabled()) {
-            return elasticsearchClient.searchLayerMemoryHistory(layerName, shows, startTime,
-                    endTime, maxResults);
-        }
-        return Collections.emptyList();
-    }
 }