L Server endpoints config

examples/server.config.example

@@ -37,8 +37,17 @@ instances {
       nanos: 0
     }
 
-    # limit for CAS total content size in bytes
-    cas_max_size_bytes: 1073741824 # 1024 * 1024 * 1024
+    cas_config: {
+      memory: {
+        # limit for CAS total content size in bytes
+        cas_max_size_bytes: 1073741824 # 1024 * 1024 * 1024
+      }
+    }
+
+    action_cache_config: {
+      # retain a local map of actionKeys and retain actions in CAS
+      delegate_cas: {}
+    }
 
     # an imposed action-key-invariant timeout used in the unspecified timeout case
     default_action_timeout: {
@@ -79,7 +88,25 @@ instances {
       nanos: 0
     }
 
-    cas_max_size_bytes: 655360 # 640k
+    cas_config: {
+      grpc: {
+        # instance name for CAS resources, default is empty
+        instance_name: "internal"
+
+        # target suitable for netty channel
+        target: "cas-host.cloud.org:5000"
+      }
+    }
+
+    action_cache_config: {
+      grpc: {
+        # instance name for AC requests, default is empty
+        instance_name: "lru-ac"
+
+        # target suitable for netty channel
+        target: "ac-host.cloud.org:5001"
+      }
+    }
 
     default_action_timeout: {
       seconds: 600

src/main/java/build/buildfarm/BUILD

@@ -6,7 +6,7 @@ package(
 
 java_library(
     name = "common",
-    srcs = glob(["common/**/*.java"]),
+    srcs = glob(["common/*.java"]),
     deps = [
         "//3rdparty/jvm/com/google/guava",
         "//3rdparty/jvm/com/google/protobuf:protobuf_java",
@@ -33,9 +33,12 @@ java_library(
     name = "server-instance",
     srcs = [
         "instance/AbstractServerInstance.java",
+        "instance/OperationsMap.java",
         "instance/TokenizableIterator.java",
     ],
     deps = [
+        ":ac",
+        ":cas",
         ":common",
         ":instance",
         "//3rdparty/jvm/com/google/guava",
@@ -56,6 +59,8 @@ java_library(
     name = "memory-instance",
     srcs = glob(["instance/memory/*.java"]),
     deps = [
+        ":ac",
+        ":cas",
         ":common",
         ":instance",
         ":server-instance",
@@ -193,6 +198,36 @@ java_library(
     ],
 )
 
+java_library(
+    name = "cas",
+    srcs = glob(["cas/*.java"]),
+    deps = [
+        ":common",
+        ":stub-instance",
+        "//3rdparty/jvm/com/google/guava",
+        "//3rdparty/jvm/com/google/protobuf:protobuf_java",
+        "//3rdparty/jvm/io/grpc:grpc_core",
+        "//3rdparty/jvm/io/grpc:grpc_netty",
+        "//src/main/protobuf:build_buildfarm_v1test_buildfarm_java_proto",
+        "@googleapis//:google_bytestream_bytestream_java_grpc",
+        "@googleapis//:google_bytestream_bytestream_java_proto",
+        "@googleapis//:google_devtools_remoteexecution_v1test_remote_execution_java_grpc",
+        "@googleapis//:google_devtools_remoteexecution_v1test_remote_execution_java_proto",
+    ],
+)
+
+java_library(
+    name = "ac",
+    srcs = glob(["ac/*.java"]),
+    deps = [
+        ":common",
+        "//3rdparty/jvm/com/google/guava",
+        "//3rdparty/jvm/io/grpc:grpc_core",
+        "@googleapis//:google_devtools_remoteexecution_v1test_remote_execution_java_grpc",
+        "@googleapis//:google_devtools_remoteexecution_v1test_remote_execution_java_proto",
+    ],
+)
+
 java_binary(
     name = "buildfarm-worker",
     main_class = "build.buildfarm.worker.operationqueue.Worker",
@@ -216,4 +251,4 @@ container_image(
         "buildfarm-worker_deploy.jar",
         "/config/worker.config",
     ],
-)
+)

src/main/java/build/buildfarm/ac/ActionCache.java

@@ -0,0 +1,24 @@
+// Copyright 2018 The Bazel Authors. All rights reserved.
+//
+// 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 build.buildfarm.ac;
+
+import build.buildfarm.common.DigestUtil.ActionKey;
+import com.google.devtools.remoteexecution.v1test.ActionResult;
+
+public interface ActionCache {
+  ActionResult get(ActionKey actionKey);
+
+  void put(ActionKey actionKey, ActionResult actionResult) throws InterruptedException;
+}

src/main/java/build/buildfarm/ac/GrpcActionCache.java

@@ -0,0 +1,61 @@
+// Copyright 2018 The Bazel Authors. All rights reserved.
+//
+// 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 build.buildfarm.ac;
+
+import build.buildfarm.common.DigestUtil.ActionKey;
+import com.google.common.base.Supplier;
+import com.google.common.base.Suppliers;
+import com.google.devtools.remoteexecution.v1test.ActionResult;
+import com.google.devtools.remoteexecution.v1test.ActionCacheGrpc;
+import com.google.devtools.remoteexecution.v1test.ActionCacheGrpc.ActionCacheBlockingStub;
+import com.google.devtools.remoteexecution.v1test.GetActionResultRequest;
+import com.google.devtools.remoteexecution.v1test.UpdateActionResultRequest;
+import io.grpc.Channel;
+
+public class GrpcActionCache implements ActionCache {
+  private final String instanceName;
+  private final Channel channel;
+
+  public GrpcActionCache(String instanceName, Channel channel) {
+    this.instanceName = instanceName;
+    this.channel = channel;
+  }
+
+  private final Supplier<ActionCacheBlockingStub> actionCacheBlockingStub =
+      Suppliers.memoize(
+          new Supplier<ActionCacheBlockingStub>() {
+            @Override
+            public ActionCacheBlockingStub get() {
+              return ActionCacheGrpc.newBlockingStub(channel);
+            }
+          });
+
+  @Override
+  public ActionResult get(ActionKey actionKey) {
+    return actionCacheBlockingStub.get().getActionResult(GetActionResultRequest.newBuilder()
+        .setInstanceName(instanceName)
+        .setActionDigest(actionKey.getDigest())
+        .build());
+  }
+
+  @Override
+  public void put(ActionKey actionKey, ActionResult actionResult) {
+    actionCacheBlockingStub.get().updateActionResult(UpdateActionResultRequest.newBuilder()
+        .setInstanceName(instanceName)
+        .setActionDigest(actionKey.getDigest())
+        .setActionResult(actionResult)
+        .build());
+  }
+}

src/main/java/build/buildfarm/common/ContentAddressableStorage.java → src/main/java/build/buildfarm/cas/ContentAddressableStorage.java

@@ -12,8 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package build.buildfarm.common;
+package build.buildfarm.cas;
 
+import build.buildfarm.common.DigestUtil;
 import build.buildfarm.common.ThreadSafety.ThreadSafe;
 import com.google.devtools.remoteexecution.v1test.Digest;
 import com.google.protobuf.ByteString;
@@ -28,8 +29,13 @@ public static final class Blob {
     private final ByteString data;
 
     public Blob(ByteString data, DigestUtil digestUtil) {
+      this.data = data;
       digest = digestUtil.compute(data);
+    }
+
+    public Blob(ByteString data, Digest digest) {
       this.data = data;
+      this.digest = digest;
     }
 
     public Digest getDigest() {
@@ -59,7 +65,7 @@ public boolean isEmpty() {
 
   /** Insert a blob into the CAS. */
   @ThreadSafe
-  void put(Blob blob);
+  void put(Blob blob) throws InterruptedException;
 
   /**
    * Insert a value into the CAS with expiration callback.
@@ -71,5 +77,5 @@ public boolean isEmpty() {
    * guaranteed for invocation.
    */
   @ThreadSafe
-  void put(Blob blob, Runnable onExpiration);
+  void put(Blob blob, Runnable onExpiration) throws InterruptedException;
 }

src/main/java/build/buildfarm/cas/ContentAddressableStorages.java

@@ -0,0 +1,52 @@
+// Copyright 2018 The Bazel Authors. All rights reserved.
+//
+// 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 build.buildfarm.cas;
+
+import build.buildfarm.instance.stub.ByteStreamUploader;
+import build.buildfarm.instance.stub.Retrier;
+import build.buildfarm.v1test.ContentAddressableStorageConfig;
+import build.buildfarm.v1test.GrpcCASConfig;
+import io.grpc.Channel;
+import io.grpc.netty.NegotiationType;
+import io.grpc.netty.NettyChannelBuilder;
+
+public final class ContentAddressableStorages {
+  private static Channel createChannel(String target) {
+    NettyChannelBuilder builder =
+        NettyChannelBuilder.forTarget(target)
+            .negotiationType(NegotiationType.PLAINTEXT);
+    return builder.build();
+  }
+
+  private static ContentAddressableStorage createGrpcCAS(GrpcCASConfig config) {
+    Channel channel = createChannel(config.getTarget());
+    ByteStreamUploader byteStreamUploader
+        = new ByteStreamUploader("", channel, null, 300, Retrier.NO_RETRIES, null);
+
+    return new GrpcCAS(config.getInstanceName(), channel, byteStreamUploader);
+  }
+
+  public static ContentAddressableStorage create(ContentAddressableStorageConfig config) {
+    switch (config.getTypeCase()) {
+      default:
+      case TYPE_NOT_SET:
+        throw new IllegalArgumentException("CAS config not set in config");
+      case GRPC:
+        return createGrpcCAS(config.getGrpc());
+      case MEMORY:
+        return new MemoryCAS(config.getMemory().getMaxSizeBytes());
+    }
+  }
+}