@SuppressWarnings("resource") @Test public void testCheckAndSetLoggingAuditHeadersForLogMessage() throws Exception{ List<LogMessage> logMessages = new ArrayList<>(); List<LoggingAuditHeaders> originalHeaders = new ArrayList<>(); int pid = new Random().nextInt(); long session = System.currentTimeMillis(); for (int i = 0; i < NUM_EVENTS; i++) { LogMessage logMessage = new LogMessage(); logMessage.setKey(ByteBuffer.allocate(100).put(String.valueOf(i).getBytes())); logMessage.setMessage(ByteBuffer.allocate(100).put(String.valueOf(i).getBytes())); LoggingAuditHeaders headers = new LoggingAuditHeaders() .setHost("host-name") .setLogName("topicx") .setPid(pid) .setSession(session) .setLogSeqNumInSession(i); originalHeaders.add(headers); logMessage.setLoggingAuditHeaders(headers); logMessages.add(logMessage); } KafkaMessagePartitioner partitioner = new Crc32ByteArrayPartitioner(); KafkaProducerConfig config = new KafkaProducerConfig(); SingerSettings.setSingerConfig(new SingerConfig()); KafkaProducerManager.injectTestProducer(config, producer); SingerLogConfig singerLogConfig1 = createSingerLogConfig("test", "/a/b/c"); SingerLog singerLog1 = new SingerLog(singerLogConfig1); LogStream logStream1 = new LogStream(singerLog1, "test.tmp"); KafkaWriter writer = new KafkaWriter(logStream1, config, partitioner, "topicx", false, Executors.newCachedThreadPool(), true); assertNull(writer.getAuditHeadersGenerator()); assertNull(writer.getAuditConfig()); assertFalse(writer.isEnableLoggingAudit()); for(int i = 0; i < logMessages.size(); i++){ writer.checkAndSetLoggingAuditHeadersForLogMessage(logMessages.get(i)); assertEquals(originalHeaders.get(i), logMessages.get(i).getLoggingAuditHeaders()); } SingerLogConfig singerLogConfig2 = createSingerLogConfig("test", "/a/b/c"); singerLogConfig2.setEnableLoggingAudit(true); singerLogConfig2.setAuditConfig(new AuditConfig().setStartAtCurrentStage(false) .setStopAtCurrentStage(false).setSamplingRate(1.0)); SingerLog singerLog2 = new SingerLog(singerLogConfig2); LogStream logStream2 = new LogStream(singerLog2, "test.tmp"); writer = new KafkaWriter(logStream2, config, partitioner, "topicx", false, Executors.newCachedThreadPool(), true); assertNotNull(writer.getAuditHeadersGenerator()); assertNotNull(writer.getAuditConfig()); assertTrue(writer.isEnableLoggingAudit()); for(int i = 0; i < logMessages.size(); i++){ writer.checkAndSetLoggingAuditHeadersForLogMessage(logMessages.get(i)); assertEquals(originalHeaders.get(i), logMessages.get(i).getLoggingAuditHeaders()); } SingerLogConfig singerLogConfig3 = createSingerLogConfig("test", "/a/b/c"); singerLogConfig3.setEnableLoggingAudit(true); singerLogConfig3.setAuditConfig(new AuditConfig().setStartAtCurrentStage(true) .setStopAtCurrentStage(false).setSamplingRate(1.0)); SingerLog singerLog3 = new SingerLog(singerLogConfig3); LogStream logStream3 = new LogStream(singerLog3, "test.tmp"); writer = new KafkaWriter(logStream3, config, partitioner, "topicx", false, Executors.newCachedThreadPool(), true); String hostName = "testHostName"; String logName = "testLogName"; writer.setAuditHeadersGenerator(new AuditHeadersGenerator(hostName, logName)); assertNotNull(writer.getAuditHeadersGenerator()); assertNotNull(writer.getAuditConfig()); assertTrue(writer.isEnableLoggingAudit()); for(int i = 0; i < logMessages.size(); i++) { writer.checkAndSetLoggingAuditHeadersForLogMessage(logMessages.get(i)); assertNotEquals(originalHeaders.get(i), logMessages.get(i).getLoggingAuditHeaders()); LoggingAuditHeaders headers = logMessages.get(i).getLoggingAuditHeaders(); assertEquals(hostName, headers.getHost()); assertEquals(logName, headers.getLogName()); assertEquals(i, headers.getLogSeqNumInSession()); } writer.close(); }
public void checkAndSetLoggingAuditHeadersForLogMessage(LogMessage msg){ if (enableLoggingAudit && auditConfig.isStartAtCurrentStage() && ThreadLocalRandom.current().nextDouble() < auditConfig.getSamplingRate()){ LoggingAuditHeaders loggingAuditHeaders = null; try { loggingAuditHeaders = auditHeadersGenerator.generateHeaders(); long checksum = computeCRC(msg.getMessage()); msg.setLoggingAuditHeaders(loggingAuditHeaders); msg.setChecksum(checksum); LOG.debug("Setting loggingAuditHeaders {} for {}", loggingAuditHeaders, logName); OpenTsdbMetricConverter.incr(SingerMetrics.AUDIT_HEADERS_SET_FOR_LOG_MESSAGE, "topic=" + topic, "host=" + HOSTNAME, "logName=" + msg.getLoggingAuditHeaders().getLogName(), "logStreamName=" + logName); } catch (Exception e){ OpenTsdbMetricConverter.incr(SingerMetrics.AUDIT_HEADERS_SET_FOR_LOG_MESSAGE_EXCEPTION, "topic=" + topic, "host=" + HOSTNAME, "logName=" + msg.getLoggingAuditHeaders().getLogName(), "logStreamName=" + logName); LOG.debug("Couldn't set loggingAuditHeaders {} for {} as logging audit is enabled " + "and start at Singer {} ", loggingAuditHeaders, logName, auditConfig); } } }
KafkaWriter implements LogStreamWriter { public void checkAndSetLoggingAuditHeadersForLogMessage(LogMessage msg){ if (enableLoggingAudit && auditConfig.isStartAtCurrentStage() && ThreadLocalRandom.current().nextDouble() < auditConfig.getSamplingRate()){ LoggingAuditHeaders loggingAuditHeaders = null; try { loggingAuditHeaders = auditHeadersGenerator.generateHeaders(); long checksum = computeCRC(msg.getMessage()); msg.setLoggingAuditHeaders(loggingAuditHeaders); msg.setChecksum(checksum); LOG.debug("Setting loggingAuditHeaders {} for {}", loggingAuditHeaders, logName); OpenTsdbMetricConverter.incr(SingerMetrics.AUDIT_HEADERS_SET_FOR_LOG_MESSAGE, "topic=" + topic, "host=" + HOSTNAME, "logName=" + msg.getLoggingAuditHeaders().getLogName(), "logStreamName=" + logName); } catch (Exception e){ OpenTsdbMetricConverter.incr(SingerMetrics.AUDIT_HEADERS_SET_FOR_LOG_MESSAGE_EXCEPTION, "topic=" + topic, "host=" + HOSTNAME, "logName=" + msg.getLoggingAuditHeaders().getLogName(), "logStreamName=" + logName); LOG.debug("Couldn't set loggingAuditHeaders {} for {} as logging audit is enabled " + "and start at Singer {} ", loggingAuditHeaders, logName, auditConfig); } } } }
KafkaWriter implements LogStreamWriter { public void checkAndSetLoggingAuditHeadersForLogMessage(LogMessage msg){ if (enableLoggingAudit && auditConfig.isStartAtCurrentStage() && ThreadLocalRandom.current().nextDouble() < auditConfig.getSamplingRate()){ LoggingAuditHeaders loggingAuditHeaders = null; try { loggingAuditHeaders = auditHeadersGenerator.generateHeaders(); long checksum = computeCRC(msg.getMessage()); msg.setLoggingAuditHeaders(loggingAuditHeaders); msg.setChecksum(checksum); LOG.debug("Setting loggingAuditHeaders {} for {}", loggingAuditHeaders, logName); OpenTsdbMetricConverter.incr(SingerMetrics.AUDIT_HEADERS_SET_FOR_LOG_MESSAGE, "topic=" + topic, "host=" + HOSTNAME, "logName=" + msg.getLoggingAuditHeaders().getLogName(), "logStreamName=" + logName); } catch (Exception e){ OpenTsdbMetricConverter.incr(SingerMetrics.AUDIT_HEADERS_SET_FOR_LOG_MESSAGE_EXCEPTION, "topic=" + topic, "host=" + HOSTNAME, "logName=" + msg.getLoggingAuditHeaders().getLogName(), "logStreamName=" + logName); LOG.debug("Couldn't set loggingAuditHeaders {} for {} as logging audit is enabled " + "and start at Singer {} ", loggingAuditHeaders, logName, auditConfig); } } } KafkaWriter(LogStream logStream,
                     KafkaProducerConfig producerConfig,
                     String topic,
                     boolean skipNoLeaderPartitions,
                     boolean auditingEnabled,
                     String auditTopic,
                     String partitionerClassName,
                     int writeTimeoutInSeconds);  KafkaWriter(LogStream logStream,
                     KafkaProducerConfig producerConfig,
                     String topic,
                     boolean skipNoLeaderPartitions,
                     boolean auditingEnabled,
                     String auditTopic,
                     String partitionerClassName,
                     int writeTimeoutInSeconds, boolean enableHeadersInjector); @VisibleForTesting protected  KafkaWriter(LogStream logStream,
                        KafkaProducerConfig producerConfig,
                        KafkaMessagePartitioner partitioner,
                        String topic,
                        boolean skipNoLeaderPartitions,
                        ExecutorService clusterThreadPool,
                        boolean enableHeadersInjector); protected  KafkaWriter(KafkaProducerConfig producerConfig,
                        KafkaMessagePartitioner partitioner,
                        String topic,
                        boolean skipNoLeaderPartitions,
                        ExecutorService clusterThreadPool); }
KafkaWriter implements LogStreamWriter { public void checkAndSetLoggingAuditHeadersForLogMessage(LogMessage msg){ if (enableLoggingAudit && auditConfig.isStartAtCurrentStage() && ThreadLocalRandom.current().nextDouble() < auditConfig.getSamplingRate()){ LoggingAuditHeaders loggingAuditHeaders = null; try { loggingAuditHeaders = auditHeadersGenerator.generateHeaders(); long checksum = computeCRC(msg.getMessage()); msg.setLoggingAuditHeaders(loggingAuditHeaders); msg.setChecksum(checksum); LOG.debug("Setting loggingAuditHeaders {} for {}", loggingAuditHeaders, logName); OpenTsdbMetricConverter.incr(SingerMetrics.AUDIT_HEADERS_SET_FOR_LOG_MESSAGE, "topic=" + topic, "host=" + HOSTNAME, "logName=" + msg.getLoggingAuditHeaders().getLogName(), "logStreamName=" + logName); } catch (Exception e){ OpenTsdbMetricConverter.incr(SingerMetrics.AUDIT_HEADERS_SET_FOR_LOG_MESSAGE_EXCEPTION, "topic=" + topic, "host=" + HOSTNAME, "logName=" + msg.getLoggingAuditHeaders().getLogName(), "logStreamName=" + logName); LOG.debug("Couldn't set loggingAuditHeaders {} for {} as logging audit is enabled " + "and start at Singer {} ", loggingAuditHeaders, logName, auditConfig); } } } KafkaWriter(LogStream logStream,
                     KafkaProducerConfig producerConfig,
                     String topic,
                     boolean skipNoLeaderPartitions,
                     boolean auditingEnabled,
                     String auditTopic,
                     String partitionerClassName,
                     int writeTimeoutInSeconds);  KafkaWriter(LogStream logStream,
                     KafkaProducerConfig producerConfig,
                     String topic,
                     boolean skipNoLeaderPartitions,
                     boolean auditingEnabled,
                     String auditTopic,
                     String partitionerClassName,
                     int writeTimeoutInSeconds, boolean enableHeadersInjector); @VisibleForTesting protected  KafkaWriter(LogStream logStream,
                        KafkaProducerConfig producerConfig,
                        KafkaMessagePartitioner partitioner,
                        String topic,
                        boolean skipNoLeaderPartitions,
                        ExecutorService clusterThreadPool,
                        boolean enableHeadersInjector); protected  KafkaWriter(KafkaProducerConfig producerConfig,
                        KafkaMessagePartitioner partitioner,
                        String topic,
                        boolean skipNoLeaderPartitions,
                        ExecutorService clusterThreadPool); boolean isEnableLoggingAudit(); AuditHeadersGenerator getAuditHeadersGenerator(); @VisibleForTesting void setAuditHeadersGenerator(
      AuditHeadersGenerator auditHeadersGenerator); AuditConfig getAuditConfig(); @Override LogStream getLogStream(); @Override boolean isAuditingEnabled(); @VisibleForTesting HeadersInjector getHeadersInjector(); @VisibleForTesting ExecutorService getClusterThreadPool(); boolean checkMessageValidAndInjectHeaders(
          LogMessage msg, Headers headers, int indexWithinTheBucket, int partitionId,
          Map<Integer, Map<Integer, LoggingAuditHeaders>> mapOfHeadersMaps,
          Map<Integer, Map<Integer, Boolean>> mapOfMessageValidMaps); void checkAndSetLoggingAuditHeadersForLogMessage(LogMessage msg); @Override void writeLogMessages(List<LogMessage> logMessages); void enqueueLoggingAuditEvents(KafkaWritingTaskResult result,
                                        Map<Integer, LoggingAuditHeaders> headersMap,
                                        Map<Integer, Boolean> messageValidMap); void close(); }
KafkaWriter implements LogStreamWriter { public void checkAndSetLoggingAuditHeadersForLogMessage(LogMessage msg){ if (enableLoggingAudit && auditConfig.isStartAtCurrentStage() && ThreadLocalRandom.current().nextDouble() < auditConfig.getSamplingRate()){ LoggingAuditHeaders loggingAuditHeaders = null; try { loggingAuditHeaders = auditHeadersGenerator.generateHeaders(); long checksum = computeCRC(msg.getMessage()); msg.setLoggingAuditHeaders(loggingAuditHeaders); msg.setChecksum(checksum); LOG.debug("Setting loggingAuditHeaders {} for {}", loggingAuditHeaders, logName); OpenTsdbMetricConverter.incr(SingerMetrics.AUDIT_HEADERS_SET_FOR_LOG_MESSAGE, "topic=" + topic, "host=" + HOSTNAME, "logName=" + msg.getLoggingAuditHeaders().getLogName(), "logStreamName=" + logName); } catch (Exception e){ OpenTsdbMetricConverter.incr(SingerMetrics.AUDIT_HEADERS_SET_FOR_LOG_MESSAGE_EXCEPTION, "topic=" + topic, "host=" + HOSTNAME, "logName=" + msg.getLoggingAuditHeaders().getLogName(), "logStreamName=" + logName); LOG.debug("Couldn't set loggingAuditHeaders {} for {} as logging audit is enabled " + "and start at Singer {} ", loggingAuditHeaders, logName, auditConfig); } } } KafkaWriter(LogStream logStream,
                     KafkaProducerConfig producerConfig,
                     String topic,
                     boolean skipNoLeaderPartitions,
                     boolean auditingEnabled,
                     String auditTopic,
                     String partitionerClassName,
                     int writeTimeoutInSeconds);  KafkaWriter(LogStream logStream,
                     KafkaProducerConfig producerConfig,
                     String topic,
                     boolean skipNoLeaderPartitions,
                     boolean auditingEnabled,
                     String auditTopic,
                     String partitionerClassName,
                     int writeTimeoutInSeconds, boolean enableHeadersInjector); @VisibleForTesting protected  KafkaWriter(LogStream logStream,
                        KafkaProducerConfig producerConfig,
                        KafkaMessagePartitioner partitioner,
                        String topic,
                        boolean skipNoLeaderPartitions,
                        ExecutorService clusterThreadPool,
                        boolean enableHeadersInjector); protected  KafkaWriter(KafkaProducerConfig producerConfig,
                        KafkaMessagePartitioner partitioner,
                        String topic,
                        boolean skipNoLeaderPartitions,
                        ExecutorService clusterThreadPool); boolean isEnableLoggingAudit(); AuditHeadersGenerator getAuditHeadersGenerator(); @VisibleForTesting void setAuditHeadersGenerator(
      AuditHeadersGenerator auditHeadersGenerator); AuditConfig getAuditConfig(); @Override LogStream getLogStream(); @Override boolean isAuditingEnabled(); @VisibleForTesting HeadersInjector getHeadersInjector(); @VisibleForTesting ExecutorService getClusterThreadPool(); boolean checkMessageValidAndInjectHeaders(
          LogMessage msg, Headers headers, int indexWithinTheBucket, int partitionId,
          Map<Integer, Map<Integer, LoggingAuditHeaders>> mapOfHeadersMaps,
          Map<Integer, Map<Integer, Boolean>> mapOfMessageValidMaps); void checkAndSetLoggingAuditHeadersForLogMessage(LogMessage msg); @Override void writeLogMessages(List<LogMessage> logMessages); void enqueueLoggingAuditEvents(KafkaWritingTaskResult result,
                                        Map<Integer, LoggingAuditHeaders> headersMap,
                                        Map<Integer, Boolean> messageValidMap); void close(); static final String HOSTNAME; }