public CloudWatchMonitor(AWSLogsClient client, boolean cwlStreamingDisabled) {
    this.logsClient = client;
    this.cwlStreamingDisabled = cwlStreamingDisabled;
    if(!CodeBuilderValidation.checkCloudWatchMonitorConfig(logsClient)) {
        latestLogs = Arrays.asList(failedConfigurationLogsMessage);
        return;
    }
    if(cwlStreamingDisabled) {
        latestLogs = Arrays.asList(streamingDisabledMessage);
    }
    lastPollTime = 0L;
}
 

void setAwsLogsClient(AWSLogsClient awsLogsClient) {
	this.awsLogsClient = awsLogsClient;
}
 

@Test(timeout = 5000)
public void testBasic() throws InterruptedException {
	CloudWatchAppender appender = new CloudWatchAppender();
	AWSLogsClient awsLogClient = createMock(AWSLogsClient.class);
	appender.setAwsLogsClient(awsLogClient);

	appender.setMaxBatchSize(1);
	appender.setRegion("region");
	final String logGroup = "pfqoejpfqe";
	appender.setLogGroup(logGroup);
	final String logStream = "pffqjfqjpoqoejpfqe";
	appender.setLogStream(logStream);
	PatternLayout layout = new PatternLayout();
	layout.setContext(new LoggerContext());
	layout.setPattern("[%thread] %level %logger{20} - %msg%n%xThrowable");
	layout.start();
	appender.setLayout(layout);

	LoggingEvent event = new LoggingEvent();
	event.setTimeStamp(System.currentTimeMillis());
	String loggerName = "name";
	event.setLoggerName(loggerName);
	Level level = Level.DEBUG;
	event.setLevel(level);
	String message = "fjpewjfpewjfpewjfepowf";
	event.setMessage(message);

	String threadName = Thread.currentThread().getName();
	final String fullMessage = "[" + threadName + "] " + level + " " + loggerName + " - " + message + "\n";

	final PutLogEventsResult result = new PutLogEventsResult();
	String sequence = "ewopjfewfj";
	result.setNextSequenceToken(sequence);
	expect(awsLogClient.putLogEvents(isA(PutLogEventsRequest.class))).andAnswer(new IAnswer<PutLogEventsResult>() {
		@Override
		public PutLogEventsResult answer() {
			PutLogEventsRequest request = (PutLogEventsRequest) getCurrentArguments()[0];
			assertEquals(logGroup, request.getLogGroupName());
			assertEquals(logStream, request.getLogStreamName());
			List<InputLogEvent> events = request.getLogEvents();
			assertEquals(1, events.size());
			assertEquals(fullMessage, events.get(0).getMessage());
			return result;
		}
	}).times(2);
	awsLogClient.shutdown();

	// =====================================

	replay(awsLogClient);
	appender.start();
	// for coverage
	appender.start();
	appender.append(event);
	Thread.sleep(10);
	appender.append(event);
	while (appender.getEventsWrittenCount() < 2) {
		Thread.sleep(10);
	}
	appender.stop();
	verify(awsLogClient);
}
 

@Test(timeout = 5000)
public void testInstanceName() throws InterruptedException {
	CloudWatchAppender appender = new CloudWatchAppender();
	AWSLogsClient awsLogClient = createMock(AWSLogsClient.class);
	appender.setAwsLogsClient(awsLogClient);

	appender.setMaxBatchSize(1);
	appender.setRegion("region");
	final String logGroup = "pfqoejpfqe";
	appender.setLogGroup(logGroup);
	String prefix = "logstream-";
	appender.setLogStream(prefix + "%instanceName");
	final String expectedLogStream = prefix + "unknown";
	PatternLayout layout = new PatternLayout();
	layout.setPattern("%msg");
	layout.start();
	appender.setLayout(layout);

	LoggingEvent event = new LoggingEvent();
	event.setTimeStamp(System.currentTimeMillis());
	event.setLoggerName("name");
	event.setLevel(Level.DEBUG);
	event.setMessage("message");

	final PutLogEventsResult result = new PutLogEventsResult();
	result.setNextSequenceToken("ewopjfewfj");
	expect(awsLogClient.putLogEvents(isA(PutLogEventsRequest.class))).andAnswer(new IAnswer<PutLogEventsResult>() {
		@Override
		public PutLogEventsResult answer() {
			PutLogEventsRequest request = (PutLogEventsRequest) getCurrentArguments()[0];
			assertEquals(logGroup, request.getLogGroupName());
			assertEquals(expectedLogStream, request.getLogStreamName());
			return result;
		}
	});
	awsLogClient.shutdown();

	// =====================================

	replay(awsLogClient);
	appender.start();
	appender.append(event);
	while (appender.getEventsWrittenCount() < 1) {
		Thread.sleep(10);
	}
	appender.stop();
	verify(awsLogClient);
}
 

@Ignore
@Test
public void testMetricFilters() {

    List<String> lines;

    PrintStream original = System.out;
    try {
        final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        System.setOut(new PrintStream(outputStream, true));

        Logger logger = LoggerFactory.getLogger("TEST-LOGGER");
        logger.info("TEST-MESSAGE");

        TestMetricSet metricSet = new TestMetricSet();
        MetricRegistry registry = new MetricRegistry();
        registry.registerAll(metricSet);

        metricSet.testCounter.inc();
        metricSet.testMeter.mark(1L);
        metricSet.testHistogram.update(1L);
        metricSet.testTimer.update(1L, TimeUnit.MINUTES);

        Slf4jReporter reporter = Slf4jReporter.forRegistry(registry)
                .markWith(MarkerFactory.getMarker("METRIC"))
                .outputTo(logger)
                .build();

        reporter.report();
        lines = Arrays.asList(outputStream.toString().split("\\n"));
    } finally {
        System.setOut(original);
    }

    assertNotNull(lines);

    AWSLogsClient client = new AWSLogsClient();

    // TODO: Loop for each kind of metric

    String metricFilterPattern =
            String.format(COMPLETE_FILTER_PATTERN_MAP.get("COUNTER"), "test.namespace/testCounter");

    TestMetricFilterRequest request = new TestMetricFilterRequest()
            .withFilterPattern(metricFilterPattern)
            .withLogEventMessages(lines);

    TestMetricFilterResult result = client.testMetricFilter(request);

    MetricFilterMatchRecord matchRecord = result.getMatches().get(0);
    assertEquals("test.namespace/testCounter", matchRecord.getExtractedValues().get("$name"));
    assertEquals("1", matchRecord.getExtractedValues().get("$count"));

    MetricFilterMatchRecord matchRecord2 = result.getMatches().get(1);
    assertEquals("test.namespace/testGauge", matchRecord2.getExtractedValues().get("$name"));
    assertEquals("42", matchRecord2.getExtractedValues().get("$value"));
}
 

public static boolean checkCloudWatchMonitorConfig(AWSLogsClient client) {
    if(client == null) {
        return false;
    }
    return true;
}
 

public AWSLogsClient getCloudWatchLogsClient() throws InvalidInputException {
    AWSLogsClient client = new AWSLogsClient(awsCredentialsProvider, getClientConfiguration());
    client.setEndpoint("https://logs." + region + getAwsClientSuffix(region));
    return client;
}