RabbitMQ collector #1742

shakuzen · 2017-09-20T14:54:14Z

This adds a RabbitMQ collector module along with its corresponding auto-configuration.

The implementation is largely inspired by the existing collector implementations. In fact, I think some of the logic and tests could be shared between them, but I did not try to tackle that in this PR.

Resolves #1614

shakuzen · 2017-09-20T14:58:17Z

...mq/src/test/java/zipkin/collector/rabbitmq/ZipkinRabbitMqCollectorAutoConfigurationTest.java

+  }
+
+  @Test
+  @Ignore


This is ignored because it can't actually connect to a RabbitMQ server at localhost and the application context fails to refresh. This test was taken from the Kafka collector tests, but Kafka doesn't fail to start if it cannot connect to Kafka, I believe. It would still be nice to test that the conditional and rest of the auto-configuration is working. Let me know what you want to do about this.

shakuzen · 2017-09-20T15:00:12Z

...q/src/test/java/zipkin/collector/rabbitmq/ZipkinRabbitMqCollectorPropertiesOverrideTest.java

+  public static Iterable<Object[]> data() {
+    return Arrays.asList(
+        parameters("addresses", "localhost:5671, localhost:5673",
+          properties -> properties.getAddresses().toString().substring(1, properties.getAddresses().toString().length() - 1),


This is a bit wonky due to the mismatch of types: the properties uses a List<String>, the builder Address[], and of course we are passing just a String via the environment/YAML but Spring converts to a list for us. We could add a converter for Address but this felt like more work than it is worth right now.

shakuzen · 2017-09-20T15:02:28Z

zipkin-collector/rabbitmq/src/test/java/zipkin/collector/rabbitmq/RabbitMqCollectorTest.java

+public class RabbitMqCollectorTest {
+
+  private static final int RABBIT_PORT = 5672;
+  private static final String RABBIT_DOCKER_IMAGE = "rabbitmq:3.6-alpine";


This tests against the latest patch version of RabbitMQ 3.6. We could probably fairly easily parameterize this if we wanted to test against multiple versions. A 3.7 release candidate was published recently, for instance.

I chose the alpine varient just because it is a smaller image

sure.. we might make a test image at some point to reduce layers needed to be downloaded

codefromthecrypt · 2017-09-20T15:18:27Z

zipkin-collector/rabbitmq/src/main/java/zipkin/collector/rabbitmq/RabbitMqWorker.java

+      final RabbitMqSpanConsumer consumer = new RabbitMqSpanConsumer(this.builder, channel);
+      channel.basicConsume(this.builder.queue, true, this.name, consumer);
+    } catch (IOException e) {
+      throw new RuntimeException("Failed to start RabbitMQ consumer " + this.name, e);


IllegalStateException?

codefromthecrypt

Looks good. nothing major pops out at me. great work!

codefromthecrypt · 2017-09-20T15:20:25Z

zipkin-server/README.md

+### RabbitMQ collector
+The [RabbitMQ collector](../zipkin-collector/rabbitmq) will be enabled when the RabbitMQ server(s) `addresses` are set.
+
+Property | Environment Variable | Description


I might suggest following conventions in the file and/or restructuring them generally

I'm not sure I understand. I copied this convention from the corresponding table under "HTTP Collector". There seems to be a lack of consistency throughout the README. The "Self-Tracing" section reverses two of the columns, for the order of: "Variable", "Property", "Description". All the Storage sections and the "Scribe Collector" only give the environment variable and a description. Is this perhaps the convention you meant?

codefromthecrypt · 2017-09-20T15:21:35Z

zipkin-collector/rabbitmq/src/main/java/zipkin/collector/rabbitmq/RabbitMqCollector.java

+      return this;
+    }
+
+    public Builder addresses(Address[] addresses) {


I'd accept a list, so that we can later use things like AutoValue (which don't allow arrays), and just convert to array at the site you need.

codefromthecrypt · 2017-09-21T10:43:19Z

I can see the inconsistency. The reason why http collector stuff is only listed in the zipkin-server readme is because the code only exists here (there is no separate module even if maybe we should have one). You'll notice a difference in how kafka etc are noted.. listing the env variables here, but linking to a table in the kafka module's readme. Make sense?

codefromthecrypt · 2017-09-21T11:32:08Z

zipkin-collector/rabbitmq/README.md

+```
+3. Save an array of spans to a file like `sample-spans.json`
+```json
+[{"traceId":"73321ecb02595c39","id":"73321ecb02595c39","name":"get","annotations":[{"timestamp":1505990621526000,"value":"sr","endpoint":{"serviceName":"service","ipv4":"127.0.0.1","port":8080}},{"timestamp":1505990621527000,"value":"ss","endpoint":{"serviceName":"service","ipv4":"127.0.0.1","port":8080}}],"binaryAnnotations":[{"key":"http.method","value":"GET","endpoint":{"serviceName":"service","ipv4":"127.0.0.1","port":8080}}]}]


how about a v2 one!

[{ "traceId": "9032b04972e475c5", "id": "9032b04972e475c5", "kind": "SERVER", "name": "get", "timestamp": 1505990621526000, "duration": 612898, "localEndpoint": { "serviceName": "brave-webmvc-example", "ipv4": "192.168.1.113" }, "remoteEndpoint": { "serviceName": "", "ipv4": "127.0.0.1", "port": 60149 }, "tags": { "error": "500 Internal Server Error", "http.path": "/a" } }]

codefromthecrypt · 2017-09-21T11:34:16Z

zipkin-collector/rabbitmq/README.md

+```
+4. Publish them using the CLI 
+```bash
+$ rabbitmqadmin publish exchange=amq.default routing_key=zipkin.spans < sample-spans.json


I think the topic-like things have always defaulted to "zipkin" If that becomes ambiguous, we can append for whatever else we do. I'd prefer consistency and something obvious to guess.

(you might suggest this consistency is violated in ES, but that's a bit different as we have no dependency link transport, even if we have dependency link storage. if we did, we'd have to append to all the others, too.

codefromthecrypt · 2017-09-21T11:36:45Z

zipkin-collector/rabbitmq/src/main/java/zipkin/collector/rabbitmq/RabbitMqCollector.java

+  public static final class Builder implements CollectorComponent.Builder {
+    Collector.Builder delegate = Collector.builder(RabbitMqCollector.class);
+    CollectorMetrics metrics = CollectorMetrics.NOOP_METRICS;
+    String queue = "zipkin.spans";


I'd prefer plain old zipkin here, as otherwise folks might accidentally suggest it in support and have that be a lie!

codefromthecrypt · 2017-09-21T11:38:28Z

tested locally. good!

This adds a RabbitMQ collector module along with its corresponding auto-configuration.

codefromthecrypt

spent an hour or so last night and found some things..

codefromthecrypt · 2017-09-23T03:10:31Z

zipkin-collector/rabbitmq/pom.xml

+    </dependency>
+
+    <dependency>
+      <groupId>org.slf4j</groupId>


is this an indirect dep of rabbit? if not, let's not use it (as it is unnecessary). use JUL instead.

slf4j-api is the RabbitMQ client library's only required dependency but we don't need to use it ourselves. I'm not sure why I did... probably just habit.

codefromthecrypt · 2017-09-23T03:11:06Z

zipkin-collector/rabbitmq/src/main/java/zipkin/collector/rabbitmq/RabbitMqCollector.java

+    final Builder builder;
+    final int concurrency;
+    final AtomicReference<CheckResult> failure = new AtomicReference<>();
+    private Connection connection;


volatile and start and close are different threads

codefromthecrypt · 2017-09-23T03:12:13Z

zipkin-collector/rabbitmq/src/main/java/zipkin/collector/rabbitmq/RabbitMqCollector.java

+
+  static final class LazyRabbitWorkers extends LazyCloseable<ExecutorService> {
+
+    final Builder builder;


careful about saving builder here as it is mutable and start is different thread than .build(). one way is to make your builder auto-value and make a copy of the builder (or just save off the params). otherwise add a TODO mentioning the edge-case mutability concern

codefromthecrypt · 2017-09-23T03:12:47Z

zipkin-collector/rabbitmq/src/main/java/zipkin/collector/rabbitmq/RabbitMqSpanConsumer.java

+        Span span = SpanDecoder.THRIFT_DECODER.readSpan(body);
+        this.collector.accept(Collections.singletonList(span), NOOP);
+      } else {
+        this.collector.acceptSpans(body, DETECTING_DECODER, NOOP);


kill all this for collector.acceptSpans as single-span encoding was a legacy of kafka specifically (we don't need to carry that here)

Done in latest force pushed commit.

codefromthecrypt · 2017-09-23T03:14:49Z

zipkin-collector/rabbitmq/src/test/java/zipkin/collector/rabbitmq/RabbitMqCollectorTest.java

+import static zipkin.TestObjects.TRACE;
+import static zipkin.collector.rabbitmq.RabbitMqCollector.convertAddresses;
+
+public class RabbitMqCollectorTest {


there is insulation and also the ability to test locally in helpers like LazyCassandra3Storage might want to do that here

This adds a RabbitMQ collector module along with its corresponding auto-configuration.

codefromthecrypt · 2017-09-23T05:44:42Z

zipkin-collector/rabbitmq/src/main/java/zipkin/collector/rabbitmq/RabbitMQCollector.java

+    }
+
+    /**
+     * Queue zipkin spans will be consumed from. Defaults to "zipkin-spans".


nit doc drift here perhaps elsewhere (default)

codefromthecrypt · 2017-09-23T05:45:06Z

zipkin-collector/rabbitmq/src/main/java/zipkin/collector/rabbitmq/RabbitMQCollector.java

+  /** Lazy creates a connection and a queue before starting consumers */
+  static final class LazyInit extends LazyCloseable<Connection> {
+    final Builder builder;
+    final AtomicReference<CheckResult> failure = new AtomicReference<>();


oops this isn't used anymore (or needed)

codefromthecrypt · 2017-09-23T05:45:16Z

zipkin-collector/rabbitmq/src/main/java/zipkin/collector/rabbitmq/RabbitMQCollector.java

+  @Override
+  public CheckResult check() {
+    try {
+      CheckResult failure = connection.failure.get();


this should check the connection, if there's a way to

codefromthecrypt · 2017-09-23T05:46:29Z

zipkin-collector/rabbitmq/src/test/java/zipkin/collector/rabbitmq/RabbitMQCollectorTest.java

+import static zipkin.TestObjects.LOTS_OF_SPANS;
+import static zipkin.collector.rabbitmq.RabbitMQCollector.builder;
+
+public class RabbitMQCollectorTest {


nit should be renamed to ITRabbitMQCollector (as it uses docker)

codefromthecrypt · 2017-09-23T05:47:02Z

zipkin-collector/rabbitmq/src/test/java/zipkin/collector/rabbitmq/RabbitMQCollectorTest.java

+    assertThat(collector.get().check().ok).isTrue();
+  }
+
+  @Test public void startFailsWithInvalidRabbitMqServer() throws Exception {


this could possibly be moved to a normal unit test or left.. no matter

codefromthecrypt · 2017-09-23T08:18:47Z

@bsideup we are stumped as to why this fails in travis.. do you have some time to help?

bsideup · 2017-09-23T08:31:45Z

Hi Adrian! I'm happy to help, but can take a look only tomorrow, I hope it's fine :)

…

On Sat, 23 Sep 2017 at 10:19, Adrian Cole ***@***.***> wrote: @bsideup <https://github.com/bsideup> we are stumped as to why this fails in travis.. do you have some time to help? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1742 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ABAIip9jVuCLMyhO9nN7r7Vx0yCIkmjJks5slL73gaJpZM4PeAF2> .

codefromthecrypt · 2017-09-23T08:36:00Z

sounds good to me! thanks, sergei On Sat, Sep 23, 2017 at 4:31 PM, Sergei Egorov <[email protected]> wrote:

…

Hi Adrian! I'm happy to help, but can take a look only tomorrow, I hope it's fine :) On Sat, 23 Sep 2017 at 10:19, Adrian Cole ***@***.***> wrote: > @bsideup <https://github.com/bsideup> we are stumped as to why this fails > in travis.. do you have some time to help? > > — > You are receiving this because you were mentioned. > Reply to this email directly, view it on GitHub > <#1742 (comment)>, > or mute the thread > <https://github.com/notifications/unsubscribe-auth/ ABAIip9jVuCLMyhO9nN7r7Vx0yCIkmjJks5slL73gaJpZM4PeAF2> > . > — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#1742 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAD616uMeGkkRIncw6TvEHbfK3WmoLMMks5slMHzgaJpZM4PeAF2> .

shakuzen · 2017-09-25T04:26:39Z

zipkin-collector/rabbitmq/src/main/java/zipkin/collector/rabbitmq/RabbitMQCollector.java

+      Connection connection;
+      try {
+        connection = builder.connectionFactory.newConnection(builder.addresses);
+        connection.createChannel().queueDeclare(builder.queue, true, false, false, null);


The Channel created here will remain open until the connection is closed, which is not a big deal but it isn't necessary and it may be asked about by operators looking at open channels on RabbitMQ.

I guess this is implicitly closed via the connection, ya?

Yes, closing the connection will close any Channels that it opened. Ref. https://www.rabbitmq.com/releases/rabbitmq-java-client/current-javadoc/com/rabbitmq/client/Connection.html#close--

shakuzen · 2017-09-25T04:34:46Z

zipkin-collector/rabbitmq/src/main/java/zipkin/collector/rabbitmq/RabbitMQCollector.java

+      for (int i = 0; i < builder.concurrency; i++) {
+        String name = RabbitMQSpanConsumer.class.getName() + i;
+        try {
+          // TODO: Is channel 1-1 with consumer? if not re-use the same channel for each consumer


Generally speaking, yes. Channels should not be shared among threads. From the Channel JavaDoc:

Channel instances must not be shared between threads. Applications should prefer using a Channel per thread instead of sharing the same Channel across multiple threads. While some operations on channels are safe to invoke concurrently, some are not and will result in incorrect frame interleaving on the wire. Sharing channels between threads will also interfere with Publisher Confirms. As such, applications need to use a Channel per thread.

thx for the ref

bsideup · 2017-09-25T05:36:28Z

zipkin-collector/rabbitmq/src/test/java/zipkin/collector/rabbitmq/LazyRabbitMQCollector.java

+    try {
+      container = new GenericContainer(image)
+        .withExposedPorts(RABBIT_PORT)
+        .waitingFor(new HostPortWaitStrategy());


no need for it, it's a default value

bsideup · 2017-09-25T05:38:52Z

zipkin-collector/rabbitmq/src/test/java/zipkin/collector/rabbitmq/LazyRabbitMQCollector.java

+    } finally {
+      if (container != null) {
+        System.out.println("Stopping docker image " + image);
+        container.stop();


compute runs once, lazily, but you call close many times (after every JUnit statement). This is why the tests are failing

@adriancole

hmm why would this pass locally!

since this is a class rule, it shouldn't run for each test, right?

The error looks like it is failing even before tests are run...

Running zipkin.collector.rabbitmq.ITRabbitMQCollector ℹ︎ Checking the system... ✔ Docker version is newer than 1.6.0 ✔ Docker environment has more than 2GB free ✔ File should be mountable ✔ Exposed port is accessible Starting docker image rabbitmq:3.6-alpine [ERROR] Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 5.925 s <<< FAILURE! - in zipkin.collector.rabbitmq.ITRabbitMQCollector [ERROR] zipkin.collector.rabbitmq.ITRabbitMQCollector Time elapsed: 5.925 s <<< ERROR! java.lang.IllegalStateException: Unable to establish connection to RabbitMQ server Caused by: java.io.IOException Caused by: com.rabbitmq.client.ShutdownSignalException: connection error Caused by: java.io.EOFException

@bsideup
For our other tests run with test containers, I see the following output before the Checking the system... sequence:

[INFO] Running zipkin.storage.mysql.ITMySQLStorage 07:07:47.617 [main] INFO o.t.d.EnvironmentAndSystemPropertyClientProviderStrategy - Found docker client settings from environment 07:07:47.633 [main] INFO o.t.d.DockerClientProviderStrategy - Found Docker environment with Environment variables, system properties and defaults. Resolved: dockerHost=unix:///var/run/docker.sock apiVersion='{UNKNOWN_VERSION}' registryUrl='https://index.docker.io/v1/' registryUsername='travis' registryPassword='null' registryEmail='null' dockerConfig='DefaultDockerClientConfig[dockerHost=unix:///var/run/docker.sock,registryUsername=travis,registryPassword=<null>,registryEmail=<null>,registryUrl=https://index.docker.io/v1/,dockerConfig=/home/travis/.docker,sslConfig=<null>,apiVersion={UNKNOWN_VERSION}]' 07:07:47.638 [main] INFO o.testcontainers.DockerClientFactory - Docker host IP address is localhost 07:07:47.780 [main] INFO o.testcontainers.DockerClientFactory - Connected to docker: Server Version: 17.03.1-ce API Version: 1.27 Operating System: Ubuntu 14.04.5 LTS Total Memory: 7479 MB ℹ︎ Checking the system... ✔ Docker version is newer than 1.6.0 ✔ Docker environment has more than 2GB free ✔ File should be mountable ✔ Exposed port is accessible

Note it is entirely missing for the RabbitMQ tests case (as you can see from the output Adrian posted above), which leads me to believe that a different Docker discovery strategy is being used, but looking at the code for the tests I can't figure out why this would be different.

codefromthecrypt · 2017-09-25T06:53:35Z

added a commit to undo the lazy stuff (still using a class rule)

bsideup · 2017-09-25T10:25:32Z

@adriancole does it work locally if you stop your local RabbitMQ and run tests with Docker? I have a feeling that some tests use local RabbitMQ (it explains why it passes locally)

codefromthecrypt · 2017-09-25T12:35:55Z

@adriancole <https://github.com/adriancole> does it work locally if you stop your local RabbitMQ and run tests with Docker? I have a feeling that some tests use local RabbitMQ (it explains why it passes locally)

very good guess. sadly after killing my rabbit still passes. I added a log statement to verify

codefromthecrypt · 2017-09-28T05:28:02Z

pushed a commit to use travis rabbitmq service (to see if local passes)

codefromthecrypt · 2017-09-28T13:27:31Z

nice work, @shakuzen!

shakuzen commented Sep 20, 2017

View reviewed changes

shakuzen mentioned this pull request Sep 20, 2017

RabbitMQ collector support #1614

Closed

2 tasks

codefromthecrypt reviewed Sep 20, 2017

View reviewed changes

codefromthecrypt approved these changes Sep 20, 2017

View reviewed changes

shakuzen force-pushed the rabbitmq-collector branch from d682051 to 59b6f10 Compare September 21, 2017 11:15

codefromthecrypt reviewed Sep 21, 2017

View reviewed changes

shakuzen force-pushed the rabbitmq-collector branch 2 times, most recently from c284fc4 to 6695045 Compare September 21, 2017 12:48

codefromthecrypt mentioned this pull request Sep 21, 2017

zipkin-stream rabbitmq exception #1743

Closed

RabbitMQ collector

0113c9e

This adds a RabbitMQ collector module along with its corresponding auto-configuration.

shakuzen force-pushed the rabbitmq-collector branch from 6695045 to 0113c9e Compare September 22, 2017 11:09

codefromthecrypt reviewed Sep 23, 2017

View reviewed changes

RabbitMQ collector

2a22de4

This adds a RabbitMQ collector module along with its corresponding auto-configuration.

shakuzen force-pushed the rabbitmq-collector branch from 3eacdbb to 2a22de4 Compare September 23, 2017 03:50

refactoring and docker deux

04b047d

codefromthecrypt reviewed Sep 23, 2017

View reviewed changes

Adrian Cole added 2 commits September 23, 2017 14:00

rename and compile thing

ea2a2b1

try newer testcontainers

32588e1

shakuzen commented Sep 25, 2017

View reviewed changes

bsideup reviewed Sep 25, 2017

View reviewed changes

Undoes lazy things

a7f44ac

moar logging

db18290

adds rabbitmq travis service

e34dd6f

Adrian Cole added 2 commits September 28, 2017 14:16

Attempt to correct fallback logic

ecddfe2

polish

600227c

codefromthecrypt merged commit e240b0b into openzipkin:master Sep 28, 2017

+                }
+                @Test
+                @Ignore


		static final class LazyRabbitWorkers extends LazyCloseable<ExecutorService> {

		final Builder builder;

RabbitMQ collector #1742

RabbitMQ collector #1742

Conversation

shakuzen commented Sep 20, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codefromthecrypt left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codefromthecrypt commented Sep 21, 2017 via email

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codefromthecrypt commented Sep 21, 2017

codefromthecrypt left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codefromthecrypt commented Sep 23, 2017

bsideup commented Sep 23, 2017 via email

codefromthecrypt commented Sep 23, 2017 via email

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codefromthecrypt commented Sep 25, 2017

bsideup commented Sep 25, 2017

codefromthecrypt commented Sep 25, 2017 via email

codefromthecrypt commented Sep 28, 2017

codefromthecrypt commented Sep 28, 2017