Add property's ttl (#48)

Author: hanahmily

CHANGES.md

@@ -8,6 +8,7 @@ Release Notes.
 ### Features
 
 * Add mod revision check to write requests
+* Add TTL to property.
 
 0.4.0
 ------------------

README.md

@@ -323,6 +323,23 @@ Property property = Property.create("default", "sw", "ui_template")
 this.client.apply(property); //created:false tagsNum:1
 ```
 
+The property supports setting TTL.
+
+```java
+Property property = Property.create("default", "sw", "temp_date")
+    .addTag(TagAndValue.newStringTag("state", "failed"))
+    .ttl("30m")
+    .build();
+this.client.apply(property); //created:false tagsNum:1 lease_id:1
+```
+The property's TTL can be extended by calling `Client.keepAliveProperty`,
+
+```java
+this.client.keepAliveProperty(1); //lease_id:1
+```
+
+The property's live time is reset to 30 minutes.
+
 ### Query
 
 Property can be queried via `Client.findProperty`,

src/main/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClient.java

@@ -544,6 +544,16 @@ public PropertyStore.DeleteResult deleteProperty(String group, String name, Stri
         return store.delete(group, name, id, tags);
     }
 
+    /**
+     * Keep alive the property
+     *
+     * @param leaseId lease id of the property
+     */
+    public void keepAliveProperty(long leaseId) throws BanyanDBException {
+        PropertyStore store = new PropertyStore(checkNotNull(this.channel));
+        store.keepAlive(leaseId);
+    }
+
     /**
      * Bind index rule to the stream
      *

src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/Property.java

@@ -19,12 +19,15 @@
 package org.apache.skywalking.banyandb.v1.client.metadata;
 
 import com.google.auto.value.AutoValue;
+import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
 import org.apache.skywalking.banyandb.model.v1.BanyandbModel;
 import org.apache.skywalking.banyandb.property.v1.BanyandbProperty;
 import org.apache.skywalking.banyandb.v1.client.TagAndValue;
+import org.apache.skywalking.banyandb.v1.client.util.IgnoreHashEquals;
 import org.apache.skywalking.banyandb.v1.client.util.TimeUtils;
 
+import javax.annotation.Nullable;
 import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.List;
@@ -35,19 +38,32 @@ public abstract class Property extends NamedSchema<BanyandbProperty.Property> {
 
     public abstract ImmutableList<TagAndValue<?>> tags();
 
+    @Nullable
+    public abstract String ttl();
+
+    @Nullable
+    @IgnoreHashEquals
+    public abstract Long leaseId();
+
     @Override
     public BanyandbProperty.Property serialize() {
         List<BanyandbModel.Tag> tags = new ArrayList<>(this.tags().size());
         for (final TagAndValue<?> tagAndValue : this.tags()) {
             tags.add(tagAndValue.build());
         }
-        return BanyandbProperty.Property.newBuilder()
+        BanyandbProperty.Property.Builder builder = BanyandbProperty.Property.newBuilder()
                 .setMetadata(BanyandbProperty.Metadata.newBuilder()
                         .setId(id())
                         .setContainer(buildMetadata())
                         .build())
-                .addAllTags(tags)
-                .build();
+                .addAllTags(tags);
+        if (!Strings.isNullOrEmpty(ttl())) {
+            builder.setTtl(ttl());
+        }
+        if (leaseId() != null) {
+            builder.setLeaseId(leaseId());
+        }
+        return builder.build();
     }
 
     public static Builder create(String group, String name, String id) {
@@ -59,6 +75,12 @@ static Property fromProtobuf(BanyandbProperty.Property pb) {
                         pb.getMetadata().getContainer().getName(),
                         pb.getMetadata().getId())
                 .setUpdatedAt(TimeUtils.parseTimestamp(pb.getUpdatedAt()));
+        if (!Strings.isNullOrEmpty(pb.getTtl())) {
+            b.setTtl(pb.getTtl());
+        }
+        if (pb.getLeaseId() > 0) {
+            b.setLeaseId(pb.getLeaseId());
+        }
 
         // build tag family spec
         for (int i = 0; i < pb.getTagsCount(); i++) {
@@ -85,6 +107,10 @@ public final Builder addTag(TagAndValue<?> tagAndValue) {
             return this;
         }
 
+        public abstract Builder setTtl(String ttl);
+
+        abstract Builder setLeaseId(long leaseId);
+
         public abstract Property build();
     }
 }

src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/PropertyStore.java

@@ -57,7 +57,7 @@ public ApplyResult apply(Property payload, Strategy strategy) throws BanyanDBExc
                 .build();
         BanyandbProperty.ApplyResponse resp = HandleExceptionsWith.callAndTranslateApiException(() ->
                 this.stub.apply(r));
-        return new AutoValue_PropertyStore_ApplyResult(resp.getCreated(), resp.getTagsNum());
+        return new AutoValue_PropertyStore_ApplyResult(resp.getCreated(), resp.getTagsNum(), resp.getLeaseId());
     }
 
     public DeleteResult delete(String group, String name, String id, String... tags) throws BanyanDBException {
@@ -115,6 +115,14 @@ public List<Property> list(String group, String name, List<String> ids, List<Str
         return resp.getPropertyList().stream().map(Property::fromProtobuf).collect(Collectors.toList());
     }
 
+    public void keepAlive(long leaseId) throws BanyanDBException {
+        BanyandbProperty.KeepAliveRequest req = BanyandbProperty.KeepAliveRequest.newBuilder()
+                .setLeaseId(leaseId)
+                .build();
+        HandleExceptionsWith.callAndTranslateApiException(() ->
+                this.stub.keepAlive(req));
+    }
+
     public enum Strategy {
         MERGE, REPLACE
     }
@@ -124,6 +132,8 @@ public abstract static class ApplyResult {
         public abstract boolean created();
 
         public abstract int tagsNum();
+
+        public abstract long leaseId();
     }
 
     @AutoValue

src/main/proto/banyandb/v1/banyandb-property.proto

@@ -34,6 +34,7 @@ message Metadata {
   string id = 2;
 }
 
+
 // Property stores the user defined data
 message Property {
   // metadata is the identity of a property
@@ -42,6 +43,13 @@ message Property {
   repeated model.v1.Tag tags = 2;
   // updated_at indicates when the property is updated
   google.protobuf.Timestamp updated_at = 3;
+  // readonly. lease_id is the ID of the lease that attached to key.
+  int64 lease_id = 4;
+  // ttl indicates the time to live of the property.
+  // It's a string in the format of "1h", "2m", "3s", "1500ms".
+  // It defaults to 0s, which means the property never expires.
+  // The minimum allowed ttl is 1s.
+  string ttl = 5;
 }
 
 message ApplyRequest {
@@ -60,6 +68,7 @@ message ApplyResponse {
   // True: the property is absent. False: the property existed.
   bool created = 1;
   uint32 tags_num = 2;
+  int64 lease_id = 3;
 }
 
 message DeleteRequest {
@@ -91,10 +100,17 @@ message ListResponse {
   repeated banyandb.property.v1.Property property = 1;
 }
 
+message KeepAliveRequest {
+  int64 lease_id = 1;
+}
+
+message KeepAliveResponse {}
+
 service PropertyService {
   // Apply creates a property if it's absent, or update a existed one based on a strategy.
   rpc Apply(ApplyRequest) returns (ApplyResponse);
   rpc Delete(DeleteRequest) returns (DeleteResponse);
   rpc Get(GetRequest) returns (GetResponse);
   rpc List(ListRequest) returns (ListResponse);
+  rpc KeepAlive(KeepAliveRequest) returns (KeepAliveResponse);
 }

src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientTestCI.java

@@ -30,7 +30,7 @@
 public class BanyanDBClientTestCI {
     private static final String REGISTRY = "ghcr.io";
     private static final String IMAGE_NAME = "apache/skywalking-banyandb";
-    private static final String TAG = "9552688d60f055f6bb539a91464e8b1db77e7309";
+    private static final String TAG = "f3abf75e66695f3c9fcae3138ad55e7a9d74f39a";
 
     private static final String IMAGE = REGISTRY + "/" + IMAGE_NAME + ":" + TAG;
 

src/test/java/org/apache/skywalking/banyandb/v1/client/ITBanyanDBPropertyTests.java

@@ -22,6 +22,7 @@
 import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException;
 import org.apache.skywalking.banyandb.v1.client.metadata.Group;
 import org.apache.skywalking.banyandb.v1.client.metadata.Property;
+import org.apache.skywalking.banyandb.v1.client.metadata.PropertyStore;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -128,4 +129,16 @@ public void test_PropertyList() throws BanyanDBException {
         });
     }
 
+    @Test
+    public void test_PropertyKeepAlive() throws BanyanDBException {
+        Property property = Property.create("default", "sw", "id1")
+                .setTtl("30m")
+                .addTag(TagAndValue.newStringTag("name", "bar"))
+                .build();
+        PropertyStore.ApplyResult resp = this.client.apply(property);
+        Assert.assertTrue(resp.created());
+        Assert.assertTrue(resp.leaseId() > 0);
+        this.client.keepAliveProperty(resp.leaseId());
+    }
+
 }

src/test/java/org/apache/skywalking/banyandb/v1/client/metadata/PropertyStoreTest.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.banyandb.v1.client.metadata;
 
+import com.google.common.base.Strings;
 import io.grpc.stub.StreamObserver;
 import org.apache.skywalking.banyandb.property.v1.BanyandbProperty;
 import org.apache.skywalking.banyandb.property.v1.PropertyServiceGrpc;
@@ -32,8 +33,10 @@
 import java.io.IOException;
 import java.time.ZonedDateTime;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import static org.mockito.AdditionalAnswers.delegatesTo;
 import static org.mockito.Mockito.mock;
@@ -43,6 +46,8 @@ public class PropertyStoreTest extends AbstractBanyanDBClientTest {
 
     private Map<String, BanyandbProperty.Property> memory;
 
+    private Set<Long> leasePool;
+
     private final PropertyServiceGrpc.PropertyServiceImplBase propertyServiceImpl = mock(PropertyServiceGrpc.PropertyServiceImplBase.class, delegatesTo(
             new PropertyServiceGrpc.PropertyServiceImplBase() {
                 @Override
@@ -53,10 +58,16 @@ public void apply(BanyandbProperty.ApplyRequest request, StreamObserver<Banyandb
                     String key = format(p.getMetadata());
                     BanyandbProperty.Property v = memory.get(key);
                     memory.put(format(p.getMetadata()), p);
+                    BanyandbProperty.ApplyResponse.Builder builder = BanyandbProperty.ApplyResponse.newBuilder().setTagsNum(p.getTagsCount());
+                    if (!Strings.isNullOrEmpty(p.getTtl())) {
+                        long leaseId = System.currentTimeMillis();
+                        leasePool.add(leaseId);
+                        builder.setLeaseId(leaseId);
+                    }
                     if (v == null) {
-                        responseObserver.onNext(BanyandbProperty.ApplyResponse.newBuilder().setCreated(true).setTagsNum(p.getTagsCount()).build());
+                        responseObserver.onNext(builder.setCreated(true).build());
                     } else {
-                        responseObserver.onNext(BanyandbProperty.ApplyResponse.newBuilder().setCreated(false).setTagsNum(p.getTagsCount()).build());
+                        responseObserver.onNext(builder.setCreated(false).build());
                     }
                     responseObserver.onCompleted();
                 }
@@ -80,12 +91,22 @@ public void list(BanyandbProperty.ListRequest request, StreamObserver<BanyandbPr
                     responseObserver.onNext(BanyandbProperty.ListResponse.newBuilder().addAllProperty(memory.values()).build());
                     responseObserver.onCompleted();
                 }
+
+                public void keepAlive(BanyandbProperty.KeepAliveRequest request, StreamObserver<BanyandbProperty.KeepAliveResponse> responseObserver) {
+                    if (!leasePool.contains(request.getLeaseId())) {
+                        responseObserver.onError(new RuntimeException("lease not found"));
+                    } else {
+                        responseObserver.onNext(BanyandbProperty.KeepAliveResponse.newBuilder().build());
+                    }
+                    responseObserver.onCompleted();
+                }
             }));
 
     @Before
     public void setUp() throws IOException {
         super.setUp(bindService(propertyServiceImpl));
         this.memory = new HashMap<>();
+        this.leasePool = new HashSet<>();
         this.store = new PropertyStore(this.channel);
     }
 
@@ -138,6 +159,18 @@ public void testPropertyStore_createAndDelete() throws BanyanDBException {
         Assert.assertEquals(0, memory.size());
     }
 
+    @Test
+    public void testPropertyStore_keepAlive() throws BanyanDBException {
+        Property property = Property.create("default", "sw", "ui_template")
+                .addTag(TagAndValue.newStringTag("name", "hello"))
+                .setTtl("30m")
+                .build();
+        PropertyStore.ApplyResult resp = this.store.apply(property);
+        Assert.assertTrue(resp.created());
+        Assert.assertTrue(resp.leaseId() > 0);
+        this.store.keepAlive(resp.leaseId());
+    }
+
     static String format(BanyandbProperty.Metadata metadata) {
         return metadata.getContainer().getGroup() + ":" + metadata.getContainer().getName() + "/" + metadata.getId();
     }