Fix TCP client to detect connection closures

[markkuhn]

Issue #, if available: #99

Description of changes:
Added an additional empty os.write() to sendMessage() to test socket connection.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[brandondahler]

Executing the reproduction steps from #99, it doesn't look like it resolves the problem specifically with half-open sockets:

$ nc -l 127.0.0.1 25888 -k
{"_aws":{"Timestamp":1659633805417,"CloudWatchMetrics":[{"Namespace":"aws-embedded-metrics","Metrics":[{"Name":"Test","Unit":"None"}],"Dimensions":[["LogGroup","ServiceName","ServiceType"]]}],"LogGroupName":"Unknown-metrics"},"LogGroup":"Unknown-metrics","ServiceName":"Unknown","Test":1.0,"ServiceType":"Unknown"}
^C
$ nc -l 127.0.0.1 25888 -k
<no output>

---

Updating the reproducing code to have an extra log message also doesn't seem to work quite as expected -- it appears to duplicate the second message on retry:

final Configuration config = EnvironmentConfigurationProvider.getConfig();
config.setEnvironmentOverride(Environments.Agent);

final Environment environment = new EnvironmentProvider()
    .resolveEnvironment()
    .get();

final MetricsLogger metricsLogger = new MetricsLogger(environment);
metricsLogger.putMetric("Test", 1.0d);
metricsLogger.flush();

System.out.println("Sleeping, restart agent socket now...");
Thread.sleep(10_000);

final MetricsLogger metricsLogger2 = new MetricsLogger(environment);
metricsLogger2.putMetric("Test2", 1.0d);
metricsLogger2.flush();

final MetricsLogger metricsLogger3 = new MetricsLogger(environment);
metricsLogger3.putMetric("Test3", 1.0d);
metricsLogger3.flush();

environment.getSink()
    .shutdown()
    .join();

Output:

$ nc -l 127.0.0.1 25888 -k
{"_aws":{"Timestamp":1659633935242,"CloudWatchMetrics":[{"Namespace":"aws-embedded-metrics","Metrics":[{"Name":"Test","Unit":"None"}],"Dimensions":[["LogGroup","ServiceName","ServiceType"]]}],"LogGroupName":"Unknown-metrics"},"LogGroup":"Unknown-metrics","ServiceName":"Unknown","Test":1.0,"ServiceType":"Unknown"}
^C
$ nc -l 127.0.0.1 25888 -k
{"_aws":{"Timestamp":1659633945311,"CloudWatchMetrics":[{"Namespace":"aws-embedded-metrics","Metrics":[{"Name":"Test2","Unit":"None"}],"Dimensions":[["LogGroup","ServiceName","ServiceType"]]}],"LogGroupName":"Unknown-metrics"},"LogGroup":"Unknown-metrics","ServiceName":"Unknown","ServiceType":"Unknown","Test2":1.0}
{"_aws":{"Timestamp":1659633945311,"CloudWatchMetrics":[{"Namespace":"aws-embedded-metrics","Metrics":[{"Name":"Test3","Unit":"None"}],"Dimensions":[["LogGroup","ServiceName","ServiceType"]]}],"LogGroupName":"Unknown-metrics"},"LogGroup":"Unknown-metrics","ServiceName":"Unknown","ServiceType":"Unknown","Test3":1.0}
{"_aws":{"Timestamp":1659633945311,"CloudWatchMetrics":[{"Namespace":"aws-embedded-metrics","Metrics":[{"Name":"Test3","Unit":"None"}],"Dimensions":[["LogGroup","ServiceName","ServiceType"]]}],"LogGroupName":"Unknown-metrics"},"LogGroup":"Unknown-metrics","ServiceName":"Unknown","ServiceType":"Unknown","Test3":1.0}

[markkuhn]

The retryQueue was interfering with sendMessageForMaxAttempts(). I have removed it and replaced it with an extra os.write(). This will write 1 byte to the stream as a way to verify the connection. If it fails, the sendMessageForMaxAttempts will take over and retry. If it passes then os.write(message.getBytes()) will proceed.

[giancarlokc]

I think this comment should include why we need this, rather what.

[brandondahler]

I pulled the latest changes and attempted to run the reproduction code again (see previous comment).

My results were as follows:

1. Two messages with 10s delay between to allow CTRL+C and re-start:

$ nc -l 127.0.0.1 25888 -k
 {"_aws":{"Timestamp":1660165747821,"CloudWatchMetrics":[{"Namespace":"aws-embedded-metrics","Metrics":[{"Name":"Test","Unit":"None"}],"Dimensions":[["LogGroup","ServiceName","ServiceType"]]}],"LogGroupName":"Unknown-metrics"},"LogGroup":"Unknown-metrics","ServiceName":"Unknown","Test":1.0,"ServiceType":"Unknown"}
^C
$ nc -l 127.0.0.1 25888 -k
<no output>

2. Three messages, 10s delay between 1st and 2nd to allow CTRL+C and re-start:

$ nc -l 127.0.0.1 25888 -k
 {"_aws":{"Timestamp":1660165887524,"CloudWatchMetrics":[{"Namespace":"aws-embedded-metrics","Metrics":[{"Name":"Test","Unit":"None"}],"Dimensions":[["LogGroup","ServiceName","ServiceType"]]}],"LogGroupName":"Unknown-metrics"},"LogGroup":"Unknown-metrics","ServiceName":"Unknown","Test":1.0,"ServiceType":"Unknown"}
^C
$ nc -l 127.0.0.1 25888 -k
 {"_aws":{"Timestamp":1660165897592,"CloudWatchMetrics":[{"Namespace":"aws-embedded-metrics","Metrics":[{"Name":"Test3","Unit":"None"}],"Dimensions":[["LogGroup","ServiceName","ServiceType"]]}],"LogGroupName":"Unknown-metrics"},"LogGroup":"Unknown-metrics","ServiceName":"Unknown","ServiceType":"Unknown","Test3":1.0}

Neither of these tests resulted in the expected behavior.

[markkuhn]

@brandondahler I ran tests again and I see now that in about 5% of the cases, the socket doesn't detect its closure in time and the message gets lost.
I would like to use your solution but after running some tests I can see that it does sometimes miss a message. At times it also sends a message twice which isn't something we can have. I'm going to do some digging to see if there's another way to approach this.