ignore events with pubsub; add protobuf serializer

agent/pom.xml

@@ -14,7 +14,7 @@
   </parent>
 
   <properties>
-    <spotify-metrics.version>0.7.0</spotify-metrics.version>
+    <spotify-metrics.version>1.0.1</spotify-metrics.version>
   </properties>
 
 

core/pom.xml

@@ -25,6 +25,11 @@
       <scope>provided</scope>
     </dependency>
 
+    <dependency>
+      <groupId>com.google.protobuf</groupId>
+      <artifactId>protobuf-java</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>eu.toolchain.async</groupId>
       <artifactId>tiny-async-core</artifactId>

core/src/main/java/com/spotify/ffwd/serializer/BuiltInSerializers.java

@@ -31,6 +31,7 @@ public class BuiltInSerializers implements FastForwardModule {
     @Override
     public void setup() throws Exception {
         context.registerSerializer("spotify100", Spotify100Serializer.class);
+        context.registerSerializer("spotify100proto", Spotify100ProtoSerializer.class);
         context.registerSerializer("to-string", ToStringSerializer.class);
     }
 }

core/src/main/java/com/spotify/ffwd/serializer/Spotify100ProtoSerializer.java

@@ -0,0 +1,48 @@
+/*-
+ * -\-\-
+ * FastForward Core
+ * --
+ * Copyright (C) 2016 - 2018 Spotify AB
+ * --
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * -/-/-
+ */
+
+package com.spotify.ffwd.serializer;
+
+import com.spotify.ffwd.model.Event;
+import com.spotify.ffwd.model.Metric;
+import com.spotify.proto.Spotify100;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class Spotify100ProtoSerializer implements Serializer {
+
+  @Override
+  public byte[] serialize(final Event event) throws Exception {
+    log.debug("Serializing events is not supported");
+    return new byte[0];
+  }
+
+  @Override
+  public byte[] serialize(final Metric metric) throws Exception {
+    return Spotify100.Metric.newBuilder()
+      .setKey(metric.getKey())
+      .setTime(metric.getTime().getTime())
+      .setValue(metric.getValue())
+      .putAllTags(metric.getTags())
+      .putAllResource(metric.getResource())
+      .build()
+      .toByteArray();
+  }
+}

core/src/main/proto/spotify_100.proto

@@ -0,0 +1,27 @@
+syntax = "proto3";
+
+package spotify.metric;
+option java_package = "com.spotify.proto";
+
+option optimize_for = SPEED;
+
+message Metric {
+    // key of metric.
+    string key = 1;
+
+    // time in ms when metric was generated.
+    int64 time = 2;
+
+    // value of metric.
+    double value = 3;
+
+    // tags associated to a metric, used to be referred to as attributes.
+    map<string, string> tags = 4;
+
+    // resource "tags" associated to metric.
+    map<string, string> resource = 5;
+}
+
+message Message {
+    Metric metric = 1;
+}

core/src/test/java/com/spotify/ffwd/serializer/TestSpotify100ProtoSerializer.java

@@ -0,0 +1,71 @@
+/*-
+ * -\-\-
+ * FastForward Core
+ * --
+ * Copyright (C) 2016 - 2018 Spotify AB
+ * --
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * -/-/-
+ */
+
+package com.spotify.ffwd.serializer;
+
+import static org.hamcrest.CoreMatchers.any;
+import static org.junit.Assert.assertThat;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import com.spotify.ffwd.model.Metric;
+import java.time.Instant;
+import java.util.Date;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestSpotify100ProtoSerializer {
+  private Spotify100ProtoSerializer spotify100ProtoSerializer;
+  private Metric metric;
+
+  @Before
+  public void setup() {
+    spotify100ProtoSerializer = new Spotify100ProtoSerializer();
+  }
+
+  @Test
+  public void testSerializeMetric() throws Exception {
+    metric = new Metric("",
+      0.0,
+      Date.from(Instant.ofEpochSecond(1542812184)),
+      ImmutableSet.of(),
+      ImmutableMap.of("a", "b"),
+      ImmutableMap.of(),
+      "");
+
+    final byte[] serialize = spotify100ProtoSerializer.serialize(metric);
+
+    assertThat(serialize, any(byte[].class));
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void testNullKeyException() throws Exception {
+    metric = new Metric(null,
+      0.0,
+      Date.from(Instant.ofEpochSecond(1542812184)),
+      ImmutableSet.of(),
+      ImmutableMap.of("a", "b"),
+      ImmutableMap.of(),
+      "");
+
+    spotify100ProtoSerializer.serialize(metric);
+  }
+
+}

modules/json/pom.xml

@@ -23,5 +23,13 @@
       <groupId>com.spotify.ffwd</groupId>
       <artifactId>ffwd-api</artifactId>
     </dependency>
+
+    <!-- testing -->
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
+
 </project>

modules/json/src/main/java/com/spotify/ffwd/json/JsonObjectMapperDecoder.java

@@ -169,10 +169,10 @@ private double decodeDouble(JsonNode tree, String name) {
         return n.asDouble();
     }
 
-    private String decodeString(JsonNode tree, String name) {
+    String decodeString(JsonNode tree, String name) {
         final JsonNode n = tree.get(name);
 
-        if (n == null) {
+        if (n.isNull()) {
             return null;
         }
 

modules/json/src/test/java/com/spotify/ffwd/json/JsonObjectMapperDecoderTest.java

@@ -0,0 +1,49 @@
+/*-
+ * -\-\-
+ * FastForward JSON Module
+ * --
+ * Copyright (C) 2016 - 2018 Spotify AB
+ * --
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * -/-/-
+ */
+
+package com.spotify.ffwd.json;
+
+import static junit.framework.TestCase.assertNull;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.IOException;
+import org.junit.Before;
+import org.junit.Test;
+
+public class JsonObjectMapperDecoderTest {
+  private final ObjectMapper mapper = new ObjectMapper();
+
+  private JsonObjectMapperDecoder j;
+
+  @Before
+  public void setup() {
+   j = new JsonObjectMapperDecoder();
+  }
+
+  @Test
+  public void nullKey() throws IOException {
+    final JsonNode json = mapper.readTree("{\"key\": null}");
+    final String key = j.decodeString(json, "key");
+
+    assertNull(key);
+  }
+
+}

modules/pubsub/src/main/java/com/spotify/ffwd/pubsub/PubsubPluginSink.java

@@ -90,16 +90,7 @@ public void init() { }
 
   @Override
   public AsyncFuture<Void> sendEvents(Collection<Event> events) {
-    for (Event event : events) {
-      try {
-        publisher.publish(PubsubMessage.newBuilder()
-          .setData(ByteString.copyFrom(serializer.serialize(event))).build()
-        );
-      } catch (Exception e) {
-        log.error("Failed to publish event {}", e);
-      }
-    }
-
+    log.debug("Sending events is not supported!");
     return async.resolved();
   }
 

pom.xml

@@ -224,6 +224,12 @@
         <version>${project.version}</version>
       </dependency>
 
+      <dependency>
+        <groupId>com.google.protobuf</groupId>
+        <artifactId>protobuf-java</artifactId>
+        <version>3.5.1</version>
+      </dependency>
+
       <dependency>
         <groupId>com.fasterxml.jackson.core</groupId>
         <artifactId>jackson-databind</artifactId>
@@ -336,7 +342,31 @@
   </dependencyManagement>
 
   <build>
+    <extensions>
+      <extension>
+        <groupId>kr.motd.maven</groupId>
+        <artifactId>os-maven-plugin</artifactId>
+        <version>1.5.0.Final</version>
+      </extension>
+    </extensions>
+
     <plugins>
+      <plugin>
+        <groupId>org.xolstice.maven.plugins</groupId>
+        <artifactId>protobuf-maven-plugin</artifactId>
+        <version>0.6.1</version>
+        <configuration>
+          <protocArtifact>com.google.protobuf:protoc:3.6.1:exe:${os.detected.classifier}</protocArtifact>
+        </configuration>
+        <executions>
+          <execution>
+            <goals>
+              <goal>compile</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+
       <plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>findbugs-maven-plugin</artifactId>