Store Notifications in ZK

3 files changed, 188 insertions, 0 deletions

diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/persistence/CuratorDb.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/persistence/CuratorDb.java
index bf1ee4b23ce..3d6cb45aeb1 100644
--- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/persistence/CuratorDb.java
+++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/persistence/CuratorDb.java
@@ -25,6 +25,7 @@ import com.yahoo.vespa.hosted.controller.deployment.Run;
 import com.yahoo.vespa.hosted.controller.deployment.Step;
 import com.yahoo.vespa.hosted.controller.dns.NameServiceQueue;
 import com.yahoo.vespa.hosted.controller.api.integration.vcmr.VespaChangeRequest;
+import com.yahoo.vespa.hosted.controller.notification.Notification;
 import com.yahoo.vespa.hosted.controller.routing.GlobalRouting;
 import com.yahoo.vespa.hosted.controller.routing.RoutingPolicy;
 import com.yahoo.vespa.hosted.controller.routing.RoutingPolicyId;
@@ -89,6 +90,7 @@ public class CuratorDb {
     private static final Path endpointCertificateRoot = root.append("applicationCertificates");
     private static final Path archiveBucketsRoot = root.append("archiveBuckets");
     private static final Path changeRequestsRoot = root.append("changeRequests");
+    private static final Path notificationsRoot = root.append("notifications");
 
     private final NodeVersionSerializer nodeVersionSerializer = new NodeVersionSerializer();
     private final VersionStatusSerializer versionStatusSerializer = new VersionStatusSerializer(nodeVersionSerializer);
@@ -208,6 +210,10 @@ public class CuratorDb {
         return curator.lock(lockRoot.append("changeRequests"), defaultLockTimeout);
     }
 
+    public Lock lockNotifications(TenantName tenantName) {
+        return curator.lock(lockRoot.append("notifications").append(tenantName.value()), defaultLockTimeout);
+    }
+
     // -------------- Helpers ------------------------------------------
 
     /** Try locking with a low timeout, meaning it is OK to fail lock acquisition.
@@ -589,6 +595,21 @@ public class CuratorDb {
         curator.delete(changeRequestPath(changeRequest.getId()));
     }
 
+    // -------------- Notifications ---------------------------------------------------
+
+    public List<Notification> readNotifications(TenantName tenantName) {
+        return readSlime(notificationsPath(tenantName))
+                .map(slime -> NotificationsSerializer.fromSlime(tenantName, slime)).orElseGet(List::of);
+    }
+
+    public void writeNotifications(TenantName tenantName, List<Notification> notifications) {
+        curator.set(notificationsPath(tenantName), asJson(NotificationsSerializer.toSlime(notifications)));
+    }
+
+    public void deleteNotifications(TenantName tenantName) {
+        curator.delete(notificationsPath(tenantName));
+    }
+
     // -------------- Paths ---------------------------------------------------
 
     private Path lockPath(TenantName tenant) {
@@ -718,4 +739,8 @@ public class CuratorDb {
         return changeRequestsRoot.append(id);
     }
 
+    private static Path notificationsPath(TenantName tenantName) {
+        return notificationsRoot.append(tenantName.value());
+    }
+
 }
diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/persistence/NotificationsSerializer.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/persistence/NotificationsSerializer.java
new file mode 100644
index 00000000000..dcb485b9016
--- /dev/null
+++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/persistence/NotificationsSerializer.java
@@ -0,0 +1,104 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.hosted.controller.persistence;
+
+import com.yahoo.config.provision.ApplicationName;
+import com.yahoo.config.provision.ClusterSpec;
+import com.yahoo.config.provision.InstanceName;
+import com.yahoo.config.provision.TenantName;
+import com.yahoo.config.provision.zone.ZoneId;
+import com.yahoo.slime.Cursor;
+import com.yahoo.slime.Inspector;
+import com.yahoo.slime.Slime;
+import com.yahoo.slime.SlimeUtils;
+import com.yahoo.vespa.hosted.controller.api.integration.deployment.JobType;
+import com.yahoo.vespa.hosted.controller.notification.Notification;
+import com.yahoo.vespa.hosted.controller.notification.NotificationSource;
+
+import java.util.List;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * (de)serializes notifications for a tenant
+ *
+ * @author freva
+ */
+public class NotificationsSerializer {
+
+    // WARNING: Since there are multiple servers in a ZooKeeper cluster and they upgrade one by one
+    //          (and rewrite all nodes on startup), changes to the serialized format must be made
+    //          such that what is serialized on version N+1 can be read by version N:
+    //          - ADDING FIELDS: Always ok
+    //          - REMOVING FIELDS: Stop reading the field first. Stop writing it on a later version.
+    //          - CHANGING THE FORMAT OF A FIELD: Don't do it bro.
+
+    private static final String notificationsFieldName = "notifications";
+    private static final String atFieldName = "at";
+    private static final String typeField = "type";
+    private static final String messagesField = "messages";
+    private static final String applicationField = "application";
+    private static final String instanceField = "instance";
+    private static final String zoneField = "zone";
+    private static final String clusterIdField = "clusterId";
+    private static final String jobTypeField = "jobId";
+    private static final String runNumberField = "runNumber";
+
+    public static Slime toSlime(List<Notification> notifications) {
+        Slime slime = new Slime();
+        Cursor notificationsArray = slime.setObject().setArray(notificationsFieldName);
+
+        for (Notification notification : notifications) {
+            Cursor notificationObject = notificationsArray.addObject();
+            notificationObject.setLong(atFieldName, notification.at().toEpochMilli());
+            notificationObject.setString(typeField, asString(notification.type()));
+            Cursor messagesArray = notificationObject.setArray(messagesField);
+            notification.messages().forEach(messagesArray::addString);
+
+            notification.source().application().ifPresent(application -> notificationObject.setString(applicationField, application.value()));
+            notification.source().instance().ifPresent(instance -> notificationObject.setString(instanceField, instance.value()));
+            notification.source().zoneId().ifPresent(zoneId -> notificationObject.setString(zoneField, zoneId.value()));
+            notification.source().clusterId().ifPresent(clusterId -> notificationObject.setString(clusterIdField, clusterId.value()));
+            notification.source().jobType().ifPresent(jobType -> notificationObject.setString(jobTypeField, jobType.jobName()));
+            notification.source().runNumber().ifPresent(runNumber -> notificationObject.setLong(runNumberField, runNumber));
+        }
+
+        return slime;
+    }
+
+    public static List<Notification> fromSlime(TenantName tenantName, Slime slime) {
+        return SlimeUtils.entriesStream(slime.get().field(notificationsFieldName))
+                .map(inspector -> fromInspector(tenantName, inspector))
+                .collect(Collectors.toUnmodifiableList());
+    }
+
+    private static Notification fromInspector(TenantName tenantName, Inspector inspector) {
+       return new Notification(
+               Serializers.instant(inspector.field(atFieldName)),
+               typeFrom(inspector.field(typeField)),
+               new NotificationSource(
+                       tenantName,
+                       Serializers.optionalString(inspector.field(applicationField)).map(ApplicationName::from),
+                       Serializers.optionalString(inspector.field(instanceField)).map(InstanceName::from),
+                       Serializers.optionalString(inspector.field(zoneField)).map(ZoneId::from),
+                       Serializers.optionalString(inspector.field(clusterIdField)).map(ClusterSpec.Id::from),
+                       Serializers.optionalString(inspector.field(jobTypeField)).map(JobType::fromJobName),
+                       Serializers.optionalLong(inspector.field(runNumberField))),
+               SlimeUtils.entriesStream(inspector.field(messagesField)).map(Inspector::asString).collect(Collectors.toUnmodifiableList()));
+    }
+    
+    private static String asString(Notification.Type type) {
+        switch (type) {
+            case APPLICATION_PACKAGE_WARNING: return "APPLICATION_PACKAGE_WARNING";
+            case DEPLOYMENT_FAILURE: return "DEPLOYMENT_FAILURE";
+            default: throw new IllegalArgumentException("No serialization defined for notification type " + type);
+        }
+    }
+
+    private static Notification.Type typeFrom(Inspector field) {
+        switch (field.asString()) {
+            case "APPLICATION_PACKAGE_WARNING": return Notification.Type.APPLICATION_PACKAGE_WARNING;
+            case "DEPLOYMENT_FAILURE": return Notification.Type.DEPLOYMENT_FAILURE;
+            default: throw new IllegalArgumentException("Unknown serialized notification type value '" + field.asString() + "'");
+        }
+    }
+}
diff --git a/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/persistence/NotificationsSerializerTest.java b/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/persistence/NotificationsSerializerTest.java
new file mode 100644
index 00000000000..f3f2d10cfd0
--- /dev/null
+++ b/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/persistence/NotificationsSerializerTest.java
@@ -0,0 +1,59 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.hosted.controller.persistence;
+
+import com.yahoo.config.provision.ApplicationId;
+import com.yahoo.config.provision.TenantName;
+import com.yahoo.slime.Slime;
+import com.yahoo.slime.SlimeUtils;
+import com.yahoo.vespa.hosted.controller.api.integration.deployment.JobType;
+import com.yahoo.vespa.hosted.controller.api.integration.deployment.RunId;
+import com.yahoo.vespa.hosted.controller.application.TenantAndApplicationId;
+import com.yahoo.vespa.hosted.controller.notification.Notification;
+import com.yahoo.vespa.hosted.controller.notification.NotificationSource;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * @author freva
+ */
+public class NotificationsSerializerTest {
+
+    @Test
+    public void serialization_test() throws IOException {
+        TenantName tenantName = TenantName.from("tenant1");
+        List<Notification> notifications = List.of(
+                new Notification(Instant.ofEpochSecond(1234),
+                        Notification.Type.APPLICATION_PACKAGE_WARNING,
+                        NotificationSource.from(TenantAndApplicationId.from(tenantName.value(), "app1")),
+                        List.of("Something something deprecated...")),
+                new Notification(Instant.ofEpochSecond(2345),
+                        Notification.Type.DEPLOYMENT_FAILURE,
+                        NotificationSource.from(new RunId(ApplicationId.from(tenantName.value(), "app1", "instance1"), JobType.systemTest, 12)),
+                        List.of("Failed to deploy: Out of capacity")));
+
+        Slime serialized = NotificationsSerializer.toSlime(notifications);
+        assertEquals("{\"notifications\":[" +
+                "{" +
+                    "\"at\":1234000," +
+                    "\"type\":\"APPLICATION_PACKAGE_WARNING\"," +
+                    "\"messages\":[\"Something something deprecated...\"]," +
+                    "\"application\":\"app1\"" +
+                "},{" +
+                    "\"at\":2345000," +
+                    "\"type\":\"DEPLOYMENT_FAILURE\"," +
+                    "\"messages\":[\"Failed to deploy: Out of capacity\"]," +
+                    "\"application\":\"app1\"," +
+                    "\"instance\":\"instance1\"," +
+                    "\"jobId\":\"system-test\"," +
+                    "\"runNumber\":12" +
+                "}]}", new String(SlimeUtils.toJsonBytes(serialized)));
+
+        List<Notification> deserialized = NotificationsSerializer.fromSlime(tenantName, serialized);
+        assertEquals(notifications, deserialized);
+    }
+}
\ No newline at end of file