Logging Service API

/**
 * HTTP server for log querying with service discovery
 * Manages lifecycle of log query endpoints and integrates with CDAP service discovery
 */
public class LogQueryService extends AbstractIdleService {
    /**
     * Starts the log query HTTP service
     * Initializes HTTP server, registers with service discovery, and begins accepting requests
     * @throws Exception if service startup fails
     */
    protected void startUp() throws Exception;
    
    /**
     * Stops the log query HTTP service  
     * Gracefully shuts down HTTP server and deregisters from service discovery
     * @throws Exception if service shutdown fails
     */
    protected void shutDown() throws Exception;
}

/**
 * REST handler providing comprehensive log retrieval API
 * Extends AbstractLogHttpHandler to provide HTTP endpoints for various log queries
 */
public class LogHttpHandler extends AbstractLogHttpHandler {
    // REST endpoints provided by this handler:
    // GET /v3/namespaces/{namespace-id}/apps/{app-id}/{program-type}/{program-id}/logs
    // GET /v3/namespaces/{namespace-id}/apps/{app-id}/{program-type}/{program-id}/runs/{run-id}/logs
    // GET /v3/namespaces/{namespace-id}/apps/{app-id}/{program-type}/{program-id}/logs/next
    // GET /v3/namespaces/{namespace-id}/apps/{app-id}/{program-type}/{program-id}/logs/prev
    // GET /v3/system/{component-id}/{service-id}/logs
    // Similar next/prev endpoints for run-specific and system logs
}

/**
 * REST handler for error classification and analysis requests
 * Provides endpoints for analyzing and categorizing program run failures
 */
public class ErrorClassificationHttpHandler extends AbstractLogHttpHandler {
    /**
     * Create error classification handler with required dependencies
     * @param accessEnforcer Security access enforcer
     * @param authenticationContext Authentication context
     * @param logReader Log reader for accessing log events
     * @param programRunFetcher Fetcher for program run records
     * @param errorLogsClassifier Classifier for analyzing error logs
     * @param cConf Configuration settings
     */
    public ErrorClassificationHttpHandler(AccessEnforcer accessEnforcer,
                                        AuthenticationContext authenticationContext,
                                        LogReader logReader,
                                        ProgramRunRecordFetcher programRunFetcher,
                                        ErrorLogsClassifier errorLogsClassifier,
                                        CConfiguration cConf);
    
    // REST Endpoint:
    // POST /v3/namespaces/{namespace-id}/apps/{app-id}/{program-type}/{program-id}/runs/{run-id}/classify - Classify program run errors
}

/**
 * Primary interface for log reading operations
 * Provides contract for reading logs with various query patterns and filtering
 */
public interface LogReader {
    /**
     * Read logs starting from a specific offset going forward in time
     * @param loggingContext Context identifying the log source (app, program, etc.)
     * @param readRange Time or offset range for reading logs
     * @param maxEvents Maximum number of log events to return
     * @param filter Filter to apply when reading logs
     * @param callback Callback to handle each log event as it's read
     */
    void getLogNext(LoggingContext loggingContext, ReadRange readRange, int maxEvents, Filter filter, Callback callback);
    
    /**
     * Read logs starting from a specific offset going backward in time
     * @param loggingContext Context identifying the log source (app, program, etc.)
     * @param readRange Time or offset range for reading logs
     * @param maxEvents Maximum number of log events to return  
     * @param filter Filter to apply when reading logs
     * @param callback Callback to handle each log event as it's read
     */
    void getLogPrev(LoggingContext loggingContext, ReadRange readRange, int maxEvents, Filter filter, Callback callback);
    
    /**
     * Read logs within a specific time range
     * @param loggingContext Context identifying the log source
     * @param fromTimeMs Start time in milliseconds since epoch
     * @param toTimeMs End time in milliseconds since epoch
     * @param filter Filter to apply when reading logs
     * @return CloseableIterator of LogEvent objects within the specified time range
     */
    CloseableIterator<LogEvent> getLog(LoggingContext loggingContext, long fromTimeMs, long toTimeMs, Filter filter);
}

/**
 * Callback contract for processing log events during reading
 * Provides lifecycle methods for handling streams of log events
 */
public interface Callback {
    /**
     * Initialize the callback before processing begins
     * Called once before any log events are processed
     */
    void init();
    
    /**
     * Handle a single log event
     * Called for each log event that matches the query criteria
     * @param logEvent The log event to process
     */
    void handle(LogEvent logEvent);
    
    /**
     * Get the current count of processed events
     * @return Number of log events processed so far
     */
    int getCount();
    
    /**
     * Clean up resources after processing completes
     * Called once after all log events have been processed
     */
    void close();
}

import io.cdap.cdap.logging.read.LogReader;
import io.cdap.cdap.logging.read.Callback;
import io.cdap.cdap.logging.read.LogEvent;
import io.cdap.cdap.logging.context.LoggingContext;

// Create a callback to handle log events
Callback logCallback = new Callback() {
    private int count = 0;
    
    @Override
    public void init() {
        System.out.println("Starting log processing...");
    }
    
    @Override  
    public void handle(LogEvent logEvent) {
        count++;
        System.out.println("Log " + count + ": " + logEvent.getLoggingEvent().getMessage());
    }
    
    @Override
    public int getCount() {
        return count;
    }
    
    @Override
    public void close() {
        System.out.println("Processed " + count + " log events");
    }
};

// Read logs using LogReader
LogReader logReader = // ... obtain LogReader instance
LoggingContext context = // ... create appropriate logging context

// Read next 100 log events
logReader.getLogNext(context, readRange, 100, Filter.EMPTY, logCallback);

// Read logs from specific time range
long fromTimeMs = System.currentTimeMillis() - 3600000; // 1 hour ago
long toTimeMs = System.currentTimeMillis();
CloseableIterator<LogEvent> logIterator = logReader.getLog(context, fromTimeMs, toTimeMs, Filter.EMPTY);

try {
    while (logIterator.hasNext()) {
        LogEvent logEvent = logIterator.next();
        // Process log event
    }
} finally {
    logIterator.close();
}

/**
 * Core data model for log events
 * Wraps ILoggingEvent with offset information for positioning
 */
public class LogEvent {
    /**
     * Get the underlying logging event
     * @return ILoggingEvent containing the actual log data
     */
    public ILoggingEvent getLoggingEvent();
    
    /**
     * Get the offset of this log event
     * @return LogOffset indicating position in the log stream
     */
    public LogOffset getOffset();
}

/**
 * Data model for individual log entries with metadata
 * Represents log entries with associated metadata for storage and retrieval
 */
public class LogData {
    /**
     * Get the log message content
     * @return String containing the log message
     */
    public String getMessage();
    
    /**
     * Get the timestamp of the log entry
     * @return Timestamp in milliseconds since epoch
     */
    public long getTimestamp();
    
    /**
     * Get the log level
     * @return Log level (DEBUG, INFO, WARN, ERROR, etc.)
     */
    public String getLevel();
}

/**
 * Structured representation of log events with formatting
 * Provides formatted log events for API responses
 */
public class FormattedLogDataEvent {
    /**
     * Get formatted log content
     * @return Formatted log event data
     */
    public Object getFormattedContent();
}

/**
 * Plain text representation of log events
 * Simple text formatting for log events
 */
public class FormattedTextLogEvent {
    /**
     * Get plain text log content
     * @return Plain text representation of the log event
     */
    public String getTextContent();
}

/**
 * Manages log buffer pipelines, recovery, cleanup, and HTTP endpoint
 * Provides high-throughput log buffering capabilities with automatic recovery
 */
public class LogBufferService extends AbstractIdleService {
    /**
     * Start log buffer service including pipeline loading and recovery
     * @throws Exception if service startup fails
     */
    protected void startUp() throws Exception;
    
    /**
     * Stop log buffer service and cleanup resources
     * @throws Exception if service shutdown fails  
     */
    protected void shutDown() throws Exception;
}

/**
 * Exception thrown when log processing pipeline configuration is invalid
 * Indicates problems with log pipeline setup or configuration
 */
public class InvalidPipelineException extends Exception {
    /**
     * Create exception with error message
     * @param message Description of the pipeline configuration error
     */
    public InvalidPipelineException(String message);
    
    /**
     * Create exception with error message and cause
     * @param message Description of the pipeline configuration error
     * @param cause Underlying exception that caused this error
     */
    public InvalidPipelineException(String message, Throwable cause);
}

/**
 * Classifies error logs and returns error classification analysis
 * Analyzes program run logs to identify error patterns and provide structured error information
 */
public class ErrorLogsClassifier {
    /**
     * Create error logs classifier with configuration
     * @param cConf Configuration containing classification rules and settings
     * @param metricsCollectionService Service for collecting classification metrics
     */
    public ErrorLogsClassifier(CConfiguration cConf, MetricsCollectionService metricsCollectionService);
    
    /**
     * Classify errors from a log iterator
     * @param logIterator Iterator of log events to analyze
     * @param runRecord Program run record containing execution details
     * @return ErrorClassificationResponse with analysis results
     */
    public ErrorClassificationResponse classify(CloseableIterator<LogEvent> logIterator, RunRecordDetail runRecord);
    
    /**
     * Check if error classification is enabled
     * @return true if classification is enabled, false otherwise
     */
    public boolean isEnabled();
}

/**
 * Rule for error classification matching
 * Defines patterns and conditions for categorizing specific error types
 */
public class ErrorClassificationRule {
    /**
     * Get the error category for this rule
     * @return ErrorCategoryEnum representing the error category
     */
    public ErrorCategoryEnum getErrorCategory();
    
    /**
     * Get error type patterns for this rule
     * @return List of error type patterns that this rule matches
     */
    public List<String> getErrorTypePatterns();
    
    /**
     * Get message patterns for this rule
     * @return List of message patterns that this rule matches
     */
    public List<String> getMessagePatterns();
}

/**
 * Response wrapper for error classification results
 * Contains classification analysis and caching information
 */
public class ErrorClassificationResponseWrapper {
    /**
     * Get the classification response
     * @return ErrorClassificationResponse with analysis results
     */
    public ErrorClassificationResponse getResponse();
    
    /**
     * Get the timestamp when classification was performed
     * @return Classification timestamp in milliseconds since epoch
     */
    public long getTimestamp();
}