[CDAP-20479] Add sidecar metadata service for credential provisioning

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/preview/DistributedPreviewManager.java

@@ -16,20 +16,25 @@
 
 package io.cdap.cdap.internal.app.preview;
 
+import com.google.common.collect.ImmutableMap;
 import com.google.inject.Inject;
 import com.google.inject.name.Named;
+import io.cdap.cdap.api.feature.FeatureFlagsProvider;
 import io.cdap.cdap.api.metrics.MetricsCollectionService;
 import io.cdap.cdap.app.preview.PreviewConfigModule;
 import io.cdap.cdap.app.preview.PreviewManager;
 import io.cdap.cdap.app.preview.PreviewRequestQueue;
 import io.cdap.cdap.app.store.preview.PreviewStore;
 import io.cdap.cdap.common.conf.CConfiguration;
 import io.cdap.cdap.common.conf.Constants;
+import io.cdap.cdap.common.conf.Constants.ArtifactLocalizer;
 import io.cdap.cdap.common.conf.SConfiguration;
+import io.cdap.cdap.common.feature.DefaultFeatureFlagsProvider;
 import io.cdap.cdap.common.utils.DirUtils;
 import io.cdap.cdap.data.runtime.DataSetsModules;
 import io.cdap.cdap.data2.dataset2.DatasetFramework;
 import io.cdap.cdap.data2.dataset2.lib.table.leveldb.LevelDBTableService;
+import io.cdap.cdap.features.Feature;
 import io.cdap.cdap.internal.app.runtime.ProgramOptionConstants;
 import io.cdap.cdap.internal.app.worker.sidecar.ArtifactLocalizerTwillRunnable;
 import io.cdap.cdap.master.spi.twill.DependentTwillPreparer;
@@ -46,6 +51,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.Writer;
+import java.net.InetAddress;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -78,6 +84,7 @@ public class DistributedPreviewManager extends DefaultPreviewManager implements
 
   private final CConfiguration cConf;
   private final Configuration hConf;
+  private final FeatureFlagsProvider featureFlagsProvider;
   private final TwillRunner twillRunner;
   private ScheduledExecutorService scheduler;
   private TwillController controller;
@@ -90,25 +97,26 @@ public class DistributedPreviewManager extends DefaultPreviewManager implements
       AccessControllerInstantiator accessControllerInstantiator,
       AccessEnforcer accessEnforcer,
       AuthenticationContext authenticationContext,
-      @Named(PreviewConfigModule.PREVIEW_LEVEL_DB) LevelDBTableService previewLevelDBTableService,
-      @Named(PreviewConfigModule.PREVIEW_CCONF) CConfiguration previewCConf,
-      @Named(PreviewConfigModule.PREVIEW_HCONF) Configuration previewHConf,
-      @Named(PreviewConfigModule.PREVIEW_SCONF) SConfiguration previewSConf,
+      @Named(PreviewConfigModule.PREVIEW_LEVEL_DB) LevelDBTableService previewLevelDbTableService,
+      @Named(PreviewConfigModule.PREVIEW_CCONF) CConfiguration previewCconf,
+      @Named(PreviewConfigModule.PREVIEW_HCONF) Configuration previewHconf,
+      @Named(PreviewConfigModule.PREVIEW_SCONF) SConfiguration previewSconf,
       PreviewRequestQueue previewRequestQueue, PreviewStore previewStore,
       PreviewRunStopper previewRunStopper, MessagingService messagingService,
       MetricsCollectionService metricsCollectionService,
       PreviewDataCleanupService previewDataCleanupService,
       TwillRunner twillRunner) {
     super(discoveryServiceClient, datasetFramework, transactionSystemClient,
         accessControllerInstantiator, accessEnforcer, authenticationContext,
-        previewLevelDBTableService,
-        previewCConf, previewHConf, previewSConf, previewRequestQueue, previewStore,
+        previewLevelDbTableService,
+        previewCconf, previewHconf, previewSconf, previewRequestQueue, previewStore,
         previewRunStopper,
         messagingService, previewDataCleanupService, metricsCollectionService);
 
     this.cConf = cConf;
     this.hConf = hConf;
     this.twillRunner = twillRunner;
+    this.featureFlagsProvider = new DefaultFeatureFlagsProvider(cConf);
   }
 
   @Override
@@ -205,6 +213,16 @@ public void run() {
               NamespaceId.SYSTEM.getNamespace());
           twillPreparer.withConfiguration(Collections.unmodifiableMap(configMap));
 
+          if (Feature.NAMESPACED_SERVICE_ACCOUNTS.isEnabled(featureFlagsProvider)) {
+            String localhost = InetAddress.getLoopbackAddress().getHostName();
+            twillPreparer = twillPreparer.withEnv(PreviewRunnerTwillRunnable.class.getSimpleName(),
+                ImmutableMap.of(
+                    ArtifactLocalizer.GCE_METADATA_HOST_ENV_VAR,
+                    String.format("%s:%s", localhost,
+                        cConf.getInt(Constants.ArtifactLocalizer.PORT))
+                ));
+          }
+
           String priorityClass = cConf.get(Constants.Preview.CONTAINER_PRIORITY_CLASS_NAME);
           if (priorityClass != null) {
             twillPreparer = twillPreparer.setSchedulerQueue(priorityClass);

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/worker/TaskWorkerServiceLauncher.java

@@ -16,12 +16,17 @@
 
 package io.cdap.cdap.internal.app.worker;
 
+import com.google.common.collect.ImmutableMap;
 import com.google.common.util.concurrent.AbstractScheduledService;
 import com.google.inject.Inject;
+import io.cdap.cdap.api.feature.FeatureFlagsProvider;
 import io.cdap.cdap.common.conf.CConfiguration;
 import io.cdap.cdap.common.conf.Constants;
+import io.cdap.cdap.common.conf.Constants.ArtifactLocalizer;
 import io.cdap.cdap.common.conf.Constants.TaskWorker;
+import io.cdap.cdap.common.feature.DefaultFeatureFlagsProvider;
 import io.cdap.cdap.common.utils.DirUtils;
+import io.cdap.cdap.features.Feature;
 import io.cdap.cdap.internal.app.runtime.ProgramOptionConstants;
 import io.cdap.cdap.internal.app.worker.sidecar.ArtifactLocalizerTwillRunnable;
 import io.cdap.cdap.master.spi.twill.DependentTwillPreparer;
@@ -35,6 +40,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.Writer;
+import java.net.InetAddress;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -64,6 +70,8 @@ public class TaskWorkerServiceLauncher extends AbstractScheduledService {
   private final CConfiguration cConf;
   private final Configuration hConf;
 
+  private final FeatureFlagsProvider featureFlagsProvider;
+
   private final TwillRunner twillRunner;
   private TwillController twillController;
 
@@ -78,6 +86,7 @@ public TaskWorkerServiceLauncher(CConfiguration cConf, Configuration hConf,
     this.cConf = cConf;
     this.hConf = hConf;
     this.twillRunner = twillRunner;
+    this.featureFlagsProvider = new DefaultFeatureFlagsProvider(cConf);
   }
 
   @Override
@@ -182,6 +191,16 @@ public void run() {
               NamespaceId.SYSTEM.getNamespace());
           twillPreparer.withConfiguration(Collections.unmodifiableMap(configMap));
 
+          if (Feature.NAMESPACED_SERVICE_ACCOUNTS.isEnabled(featureFlagsProvider)) {
+            String localhost = InetAddress.getLoopbackAddress().getHostName();
+            twillPreparer = twillPreparer.withEnv(TaskWorkerTwillRunnable.class.getSimpleName(),
+                ImmutableMap.of(
+                    ArtifactLocalizer.GCE_METADATA_HOST_ENV_VAR,
+                    String.format("%s:%s", localhost,
+                        cConf.getInt(Constants.ArtifactLocalizer.PORT))
+                ));
+          }
+
           String priorityClass = cConf.get(Constants.TaskWorker.CONTAINER_PRIORITY_CLASS_NAME);
           if (priorityClass != null) {
             twillPreparer = twillPreparer.setSchedulerQueue(priorityClass);

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/worker/sidecar/ArtifactLocalizerService.java

@@ -58,7 +58,8 @@ public class ArtifactLocalizerService extends AbstractIdleService {
         .setPort(cConf.getInt(Constants.ArtifactLocalizer.PORT))
         .setBossThreadPoolSize(cConf.getInt(Constants.ArtifactLocalizer.BOSS_THREADS))
         .setWorkerThreadPoolSize(cConf.getInt(Constants.ArtifactLocalizer.WORKER_THREADS))
-        .setHttpHandlers(new ArtifactLocalizerHttpHandlerInternal(artifactLocalizer))
+        .setHttpHandlers(new ArtifactLocalizerHttpHandlerInternal(artifactLocalizer),
+            new GcpMetadataHttpHandlerInternal(cConf))
         .build();
 
     this.cacheCleanupInterval = cConf.getInt(

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/worker/sidecar/GcpMetadataHttpHandlerInternal.java

@@ -0,0 +1,194 @@
+/*
+ * Copyright © 2023 Cask Data, Inc.
+ *
+ * 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 io.cdap.cdap.internal.app.worker.sidecar;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonSyntaxException;
+import com.google.inject.Singleton;
+import io.cdap.cdap.common.BadRequestException;
+import io.cdap.cdap.common.ForbiddenException;
+import io.cdap.cdap.common.conf.CConfiguration;
+import io.cdap.cdap.common.conf.Constants;
+import io.cdap.cdap.gateway.handlers.util.AbstractAppFabricHttpHandler;
+import io.cdap.cdap.proto.BasicThrowable;
+import io.cdap.cdap.proto.codec.BasicThrowableCodec;
+import io.cdap.cdap.proto.security.GcpMetadataTaskContext;
+import io.cdap.common.http.HttpRequests;
+import io.cdap.common.http.HttpResponse;
+import io.cdap.http.HttpHandler;
+import io.cdap.http.HttpResponder;
+import io.netty.handler.codec.http.DefaultHttpHeaders;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import java.net.URL;
+import javax.ws.rs.DELETE;
+import javax.ws.rs.GET;
+import javax.ws.rs.PUT;
+import javax.ws.rs.Path;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.HttpHeaders;
+import joptsimple.internal.Strings;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Internal {@link HttpHandler} for Artifact Localizer.
+ */
+@Singleton
+@Path("/")
+public class GcpMetadataHttpHandlerInternal extends AbstractAppFabricHttpHandler {
+
+  protected static final String METADATA_FLAVOR_HEADER_KEY = "Metadata-Flavor";
+  protected static final String METADATA_FLAVOR_HEADER_VALUE = "Google";
+  private static final Logger LOG = LoggerFactory.getLogger(GcpMetadataHttpHandlerInternal.class);
+  private static final Gson GSON = new GsonBuilder().registerTypeAdapter(BasicThrowable.class,
+      new BasicThrowableCodec()).create();
+  private final CConfiguration cConf;
+  private final String metadataServiceEndpoint;
+  private GcpMetadataTaskContext gcpMetadataTaskContext;
+
+  /**
+   * Constructs the {@link GcpMetadataHttpHandlerInternal}.
+   *
+   * @param cConf CConfiguration
+   */
+  public GcpMetadataHttpHandlerInternal(CConfiguration cConf) {
+    this.cConf = cConf;
+    this.metadataServiceEndpoint = cConf.get(
+        Constants.TaskWorker.METADATA_SERVICE_END_POINT);
+  }
+
+  /**
+   * Returns the status of metadata server.
+   *
+   * @param request The {@link HttpRequest}.
+   * @param responder a {@link HttpResponder} for sending response.
+   * @throws Exception if there is any error.
+   */
+  @GET
+  @Path("/")
+  public void status(HttpRequest request, HttpResponder responder) throws Exception {
+
+    // check that metadata header is present in the request.
+    if (!request.headers().contains(METADATA_FLAVOR_HEADER_KEY,
+        METADATA_FLAVOR_HEADER_VALUE, true)) {
+      throw new ForbiddenException(
+          String.format("Request is missing required %s header. To access the metadata server, "
+              + "you must add the %s: %s header to your request.", METADATA_FLAVOR_HEADER_KEY,
+              METADATA_FLAVOR_HEADER_KEY, METADATA_FLAVOR_HEADER_VALUE));
+    }
+    responder.sendStatus(HttpResponseStatus.OK,
+        new DefaultHttpHeaders().add(METADATA_FLAVOR_HEADER_KEY, METADATA_FLAVOR_HEADER_VALUE));
+  }
+
+  /**
+   * Returns the token of metadata server.
+   *
+   * @param request The {@link HttpRequest}.
+   * @param responder a {@link HttpResponder} for sending response.
+   * @throws Exception if there is any error.
+   */
+  @GET
+  @Path("/computeMetadata/v1/instance/service-accounts/default/token")
+  public void token(HttpRequest request, HttpResponder responder,
+      @QueryParam("scopes") String scopes) throws Exception {
+
+    LOG.debug("Token requested for namespace: {}", gcpMetadataTaskContext.getNamespace());
+    // check that metadata header is present in the request.
+    if (!request.headers().contains(METADATA_FLAVOR_HEADER_KEY,
+        METADATA_FLAVOR_HEADER_VALUE, true)) {
+      throw new ForbiddenException(
+          String.format("Request is missing required %s header. To access the metadata server, "
+                  + "you must add the %s: %s header to your request.", METADATA_FLAVOR_HEADER_KEY,
+              METADATA_FLAVOR_HEADER_KEY, METADATA_FLAVOR_HEADER_VALUE));
+    }
+
+    // TODO: CDAP-20750
+    if (metadataServiceEndpoint == null) {
+      responder.sendString(HttpResponseStatus.NOT_IMPLEMENTED,
+          String.format("%s has not been set",
+              Constants.TaskWorker.METADATA_SERVICE_END_POINT));
+      return;
+    }
+
+    try {
+      URL url = new URL(metadataServiceEndpoint);
+      if (!Strings.isNullOrEmpty(scopes)) {
+        url = new URL(String.format("%s?scopes=%s", metadataServiceEndpoint, scopes));
+      }
+      io.cdap.common.http.HttpRequest tokenRequest = io.cdap.common.http.HttpRequest.get(url)
+          .addHeader(METADATA_FLAVOR_HEADER_KEY, METADATA_FLAVOR_HEADER_VALUE)
+          .build();
+      HttpResponse tokenResponse = HttpRequests.execute(tokenRequest);
+      responder.sendJson(HttpResponseStatus.OK, tokenResponse.getResponseBodyAsString());
+    } catch (Exception ex) {
+      LOG.warn("Failed to fetch token from metadata service", ex);
+      responder.sendString(HttpResponseStatus.INTERNAL_SERVER_ERROR, exceptionToJson(ex),
+          new DefaultHttpHeaders().set(HttpHeaders.CONTENT_TYPE, "application/json"));
+    }
+  }
+
+  /**
+   * Sets the CDAP Namespace information.
+   *
+   * @param request The {@link HttpRequest}.
+   * @param responder a {@link HttpResponder} for sending response.
+   */
+  @PUT
+  @Path("/set-context")
+  public void setContext(FullHttpRequest request, HttpResponder responder)
+      throws BadRequestException {
+    this.gcpMetadataTaskContext = getGcpMetadataTaskContext(request);
+    responder.sendJson(HttpResponseStatus.OK,
+        String.format("Context was set successfully with namespace '%s'.",
+            gcpMetadataTaskContext.getNamespace()));
+  }
+
+  /**
+   * Clears the CDAP Namespace information.
+   *
+   * @param request The {@link HttpRequest}.
+   * @param responder a {@link HttpResponder} for sending response.
+   */
+  @DELETE
+  @Path("/clear-context")
+  public void clearContext(HttpRequest request, HttpResponder responder) {
+    this.gcpMetadataTaskContext = null;
+    LOG.debug("Context cleared.");
+    responder.sendStatus(HttpResponseStatus.OK);
+  }
+
+  /**
+   * Return json representation of an exception. Used to propagate exception across network for
+   * better surfacing errors and debuggability.
+   */
+  private String exceptionToJson(Exception ex) {
+    BasicThrowable basicThrowable = new BasicThrowable(ex);
+    return GSON.toJson(basicThrowable);
+  }
+
+  private GcpMetadataTaskContext getGcpMetadataTaskContext(FullHttpRequest httpRequest)
+      throws BadRequestException {
+    try {
+      return parseBody(httpRequest, GcpMetadataTaskContext.class);
+    } catch (JsonSyntaxException e) {
+      throw new BadRequestException("Invalid json object provided in request body.");
+    }
+  }
+}