WIP : Preview Runner Push based model

cdap-app-fabric/src/main/java/io/cdap/cdap/app/preview/DefaultPreviewRunnerManager.java

@@ -30,15 +30,17 @@
 import io.cdap.cdap.api.security.store.SecureStore;
 import io.cdap.cdap.app.guice.AuditLogWriterModule;
 import io.cdap.cdap.app.guice.ProgramRunnerRuntimeModule;
-import io.cdap.cdap.common.NotFoundException;
 import io.cdap.cdap.common.conf.CConfiguration;
 import io.cdap.cdap.common.conf.Constants;
 import io.cdap.cdap.common.conf.SConfiguration;
+import io.cdap.cdap.common.discovery.ResolvingDiscoverable;
+import io.cdap.cdap.common.discovery.URIScheme;
 import io.cdap.cdap.common.encryption.guice.UserCredentialAeadEncryptionModule;
 import io.cdap.cdap.common.guice.ConfigModule;
 import io.cdap.cdap.common.guice.IOModule;
 import io.cdap.cdap.common.guice.RemoteAuthenticatorModules;
 import io.cdap.cdap.common.guice.preview.PreviewDiscoveryRuntimeModule;
+import io.cdap.cdap.common.security.HttpsEnabler;
 import io.cdap.cdap.common.utils.Networks;
 import io.cdap.cdap.config.guice.ConfigStoreModule;
 import io.cdap.cdap.data.runtime.DataSetServiceModules;
@@ -48,26 +50,30 @@
 import io.cdap.cdap.data2.dataset2.lib.table.leveldb.LevelDBTableService;
 import io.cdap.cdap.data2.metadata.writer.MetadataServiceClient;
 import io.cdap.cdap.data2.metadata.writer.NoOpMetadataServiceClient;
-import io.cdap.cdap.internal.app.preview.PreviewRunnerService;
+import io.cdap.cdap.internal.app.preview.PreviewRequestPollerInfoProvider;
+import io.cdap.cdap.internal.app.preview.PreviewRunnerHttpHandlerInternal;
 import io.cdap.cdap.internal.provision.ProvisionerModule;
 import io.cdap.cdap.logging.appender.LogAppender;
 import io.cdap.cdap.logging.appender.tms.PreviewTMSLogAppender;
 import io.cdap.cdap.messaging.guice.MessagingServerRuntimeModule;
 import io.cdap.cdap.metadata.MetadataReaderWriterModules;
 import io.cdap.cdap.metrics.guice.MetricsClientRuntimeModule;
-import io.cdap.cdap.proto.id.ApplicationId;
+import io.cdap.cdap.proto.id.ProgramId;
 import io.cdap.cdap.security.auth.context.AuthenticationContextModules;
 import io.cdap.cdap.security.guice.CoreSecurityRuntimeModule;
 import io.cdap.cdap.security.guice.preview.PreviewSecureStoreModule;
+import io.cdap.http.ChannelPipelineModifier;
+import io.cdap.http.NettyHttpService;
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.http.HttpContentDecompressor;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.tephra.TransactionSystemClient;
-import org.apache.twill.common.Threads;
+import org.apache.twill.common.Cancellable;
+import org.apache.twill.discovery.DiscoveryService;
 import org.apache.twill.discovery.DiscoveryServiceClient;
-import org.apache.twill.internal.ServiceListenerAdapter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -79,43 +85,43 @@
 
   private static final Logger LOG = LoggerFactory.getLogger(DefaultPreviewRunnerManager.class);
 
   private final CConfiguration previewCConf;
   private final Configuration previewHConf;
   private final SConfiguration previewSConf;
-  private final int maxConcurrentPreviews;
   private final DiscoveryServiceClient discoveryServiceClient;
   private final DatasetFramework datasetFramework;
   private final SecureStore secureStore;
   private final TransactionSystemClient transactionSystemClient;
   private final PreviewRunnerModule previewRunnerModule;
-  private final Set<PreviewRunnerService> previewRunnerServices;
   private final LevelDBTableService previewLevelDBTableService;
-  private final PreviewRunnerServiceFactory previewRunnerServiceFactory;
+  private final PreviewRequestPollerInfoProvider pollerInfoProvider;
+  private final DiscoveryService discoveryService;
+  private NettyHttpService previewRunnerHttpService;
+  private Cancellable cancelDiscovery;
   private PreviewRunner runner;
 
   @Inject
   DefaultPreviewRunnerManager(@Named(PreviewConfigModule.PREVIEW_CCONF) CConfiguration previewCConf,
       @Named(PreviewConfigModule.PREVIEW_HCONF) Configuration previewHConf,
       @Named(PreviewConfigModule.PREVIEW_SCONF) SConfiguration previewSConf,
       SecureStore secureStore,
       DiscoveryServiceClient discoveryServiceClient,
       @Named(DataSetsModules.BASE_DATASET_FRAMEWORK) DatasetFramework datasetFramework,
       TransactionSystemClient transactionSystemClient,
       PreviewRunnerModule previewRunnerModule,
       @Named(PreviewConfigModule.PREVIEW_LEVEL_DB) LevelDBTableService previewLevelDBService,
-      PreviewRunnerServiceFactory previewRunnerServiceFactory) {
+      PreviewRequestPollerInfoProvider pollerInfoProvider, DiscoveryService discoveryService) {
     this.previewCConf = previewCConf;
     this.previewHConf = previewHConf;
     this.previewSConf = previewSConf;
     this.datasetFramework = datasetFramework;
     this.secureStore = secureStore;
     this.discoveryServiceClient = discoveryServiceClient;
     this.transactionSystemClient = transactionSystemClient;
-    this.maxConcurrentPreviews = previewCConf.getInt(Constants.Preview.POLLER_COUNT);
-    this.previewRunnerServices = ConcurrentHashMap.newKeySet();
     this.previewRunnerModule = previewRunnerModule;
     this.previewLevelDBTableService = previewLevelDBService;
-    this.previewRunnerServiceFactory = previewRunnerServiceFactory;
+    this.pollerInfoProvider = pollerInfoProvider;
+    this.discoveryService = discoveryService;
   }
 
   @Override
@@ -127,19 +133,43 @@
       ((Service) runner).startAndWait();
     }
 
-    // Create and start the preview poller services.
-    for (int i = 0; i < maxConcurrentPreviews; i++) {
-      createPreviewRunnerService().startAndWait();
+    NettyHttpService.Builder builder = NettyHttpService.builder(
+            //TODO(sidhdirenge) : Use cConf for this host address and port.
+            Constants.Service.PREVIEW_RUNNER).setHost("0.0.0.0").setPort(0)
+        .setExecThreadPoolSize(previewCConf.getInt(Constants.Preview.EXEC_THREADS))
+        .setBossThreadPoolSize(previewCConf.getInt(Constants.Preview.BOSS_THREADS))
+        .setWorkerThreadPoolSize(previewCConf.getInt(Constants.Preview.WORKER_THREADS))
+        .setChannelPipelineModifier(new ChannelPipelineModifier() {
+          @Override
+          public void modify(ChannelPipeline pipeline) {
+            pipeline.addAfter("compressor", "decompressor", new HttpContentDecompressor());
+          }
+        }).setHttpHandlers(
+            new PreviewRunnerHttpHandlerInternal(previewCConf, pollerInfoProvider, runner,
+                this::stopPreview)
+        );
+
+    if (previewCConf.getBoolean(Constants.Security.SSL.INTERNAL_ENABLED)) {
+      new HttpsEnabler().configureKeyStore(previewCConf, previewSConf).enable(builder);
     }
+    this.previewRunnerHttpService = builder.build();
+
+    LOG.debug("Starting PreviewRunnerHttpService");
+    previewRunnerHttpService.start();
+    cancelDiscovery = discoveryService.register(ResolvingDiscoverable.of(
+        URIScheme.createDiscoverable(Constants.Service.PREVIEW_RUNNER, previewRunnerHttpService)));
+    LOG.debug("Starting PreviewRunnerHttpService has completed");
   }
 
   @Override
   protected void shutDown() throws Exception {
-    // Should stop the polling service, hence individual preview runs, before stopping the top level preview runner.
-    previewRunnerServices.forEach(this::stopQuietly);
     if (runner instanceof Service) {
       stopQuietly((Service) runner);
     }
+    if (previewRunnerHttpService != null) {
+      previewRunnerHttpService.stop(1, 2, TimeUnit.SECONDS);
+    }
+    cancelDiscovery.cancel();
   }
 
   private void stopQuietly(Service service) {
@@ -151,20 +181,14 @@
   }
 
   @Override
-  public void stop(ApplicationId preview) throws Exception {
-    PreviewRunnerService runnerService = previewRunnerServices.stream()
-        .filter(r -> r.getPreviewApplication().filter(preview::equals).isPresent())
-        .findFirst()
-        .orElse(null);
-
-    if (runnerService == null) {
-      throw new NotFoundException(
-          "Preview run cannot be stopped. Please try stopping again or start new preview run.");
-    }
+  public void stop(ProgramId preview) throws Exception {
+    LOG.info("sidhdirenge - Stop called for preview {}", preview.getParent());
+    runner.stopPreview(preview);
+  }
 
-    PreviewRunnerService newRunnerService = createPreviewRunnerService();
-    runnerService.stopAndWait();
-    newRunnerService.startAndWait();
+  private void stopPreview(ProgramId program) {
+    LOG.info("sidhdirenge - stopPreview called for preview {}", program.getApplication());
+    stop();
   }
 
   /**
@@ -223,31 +247,4 @@
         }
     );
   }
-
-  /**
-   * Creates a {@link PreviewRunnerService}. It will automatically added to and removed from the
-   * {@link #previewRunnerServices} set.
-   */
-  private PreviewRunnerService createPreviewRunnerService() {
-    PreviewRunnerService previewRunnerService = previewRunnerServiceFactory.create(runner);
-
-    previewRunnerService.addListener(new ServiceListenerAdapter() {
-
-      @Override
-      public void terminated(State from) {
-        previewRunnerServices.remove(previewRunnerService);
-        if (previewRunnerServices.isEmpty()) {
-          try {
-            stop();
-          } catch (Exception e) {
-            // should not happen
-            LOG.error("Failed to shutdown the preview runner manager service.", e);
-          }
-        }
-      }
-    }, Threads.SAME_THREAD_EXECUTOR);
-
-    previewRunnerServices.add(previewRunnerService);
-    return previewRunnerService;
-  }
 }

cdap-app-fabric/src/main/java/io/cdap/cdap/app/preview/PreviewManager.java

@@ -13,7 +13,7 @@
  * License for the specific language governing permissions and limitations under
  * the License.
  */
 package io.cdap.cdap.app.preview;
 
 import com.google.gson.JsonElement;
 import io.cdap.cdap.api.security.AccessException;
@@ -102,5 +102,6 @@
    * @param pollerInfo information about the poller
    * @return {@code PreviewRequest} if such request is available in the queue
    */
+  @Deprecated
   Optional<PreviewRequest> poll(@Nullable byte[] pollerInfo);
 }

cdap-app-fabric/src/main/java/io/cdap/cdap/app/preview/PreviewManagerModule.java

@@ -31,12 +31,12 @@
 import io.cdap.cdap.gateway.handlers.CommonHandlers;
 import io.cdap.cdap.gateway.handlers.preview.PreviewErrorClassificationHttpHandler;
 import io.cdap.cdap.gateway.handlers.preview.PreviewHttpHandler;
-import io.cdap.cdap.gateway.handlers.preview.PreviewHttpHandlerInternal;
 import io.cdap.cdap.internal.app.preview.DefaultPreviewManager;
 import io.cdap.cdap.internal.app.preview.DefaultPreviewRequestQueue;
 import io.cdap.cdap.internal.app.preview.DistributedPreviewManager;
 import io.cdap.cdap.internal.app.preview.DistributedPreviewRunStopper;
 import io.cdap.cdap.internal.app.preview.PreviewDataCleanupService;
+import io.cdap.cdap.internal.app.preview.PreviewRequestPollerService;
 import io.cdap.cdap.internal.app.preview.PreviewRunStopper;
 import io.cdap.cdap.internal.app.store.preview.DefaultPreviewStore;
 import io.cdap.cdap.messaging.server.FetchHandler;
@@ -72,6 +72,7 @@ protected void configure() {
     expose(PreviewRequestQueue.class);
 
     bind(PreviewDataCleanupService.class).in(Scopes.SINGLETON);
+    bind(PreviewRequestPollerService.class).in(Scopes.SINGLETON);
 
     if (distributedRunner) {
       bind(PreviewManager.class).to(DistributedPreviewManager.class).in(Scopes.SINGLETON);
@@ -83,7 +84,6 @@ protected void configure() {
     Multibinder<HttpHandler> handlerBinder = Multibinder.newSetBinder(binder(), HttpHandler.class);
     handlerBinder.addBinding().to(PreviewHttpHandler.class);
     handlerBinder.addBinding().to(PreviewErrorClassificationHttpHandler.class);
-    handlerBinder.addBinding().to(PreviewHttpHandlerInternal.class);
 
     if (!cConf.getBoolean(MessagingSystem.MESSAGING_SERVICE_ENABLED)) {
       // Add these handlers only if messaging service endpoint doesn't exist and preview runners need to

cdap-app-fabric/src/main/java/io/cdap/cdap/app/preview/PreviewRequestQueue.java

@@ -37,6 +37,8 @@ public interface PreviewRequestQueue {
    */
   Optional<PreviewRequest> poll(@Nullable byte[] pollerInfo);
 
+  Optional<PreviewRequest> poll();
+
   /**
    * Add a preview request in the queue.
    *

cdap-app-fabric/src/main/java/io/cdap/cdap/app/preview/PreviewRunnerManagerModule.java

@@ -19,7 +19,6 @@
 import com.google.inject.Module;
 import com.google.inject.PrivateModule;
 import com.google.inject.Scopes;
-import com.google.inject.assistedinject.FactoryModuleBuilder;
 import com.google.inject.name.Names;
 import io.cdap.cdap.common.runtime.RuntimeModule;
 import io.cdap.cdap.data.runtime.DataSetsModules;
@@ -29,8 +28,8 @@
 import io.cdap.cdap.data2.dataset2.DefaultDatasetDefinitionRegistryFactory;
 import io.cdap.cdap.internal.app.preview.DirectPreviewRequestFetcher;
 import io.cdap.cdap.internal.app.preview.PreviewRequestFetcher;
+import io.cdap.cdap.internal.app.preview.PreviewRequestPollerInfoProvider;
 import io.cdap.cdap.internal.app.preview.PreviewRunStopper;
-import io.cdap.cdap.internal.app.preview.PreviewRunnerService;
 import io.cdap.cdap.internal.app.preview.RemotePreviewRequestFetcher;
 
 /**
@@ -58,16 +57,13 @@ protected void configure() {
 
         bind(PreviewRequestFetcher.class).to(DirectPreviewRequestFetcher.class)
             .in(Scopes.SINGLETON);
+        bind(PreviewRequestPollerInfoProvider.class).toInstance(() -> new byte[0]);
 
         bind(DefaultPreviewRunnerManager.class).in(Scopes.SINGLETON);
         bind(PreviewRunStopper.class).to(DefaultPreviewRunnerManager.class);
         expose(PreviewRunStopper.class);
         bind(PreviewRunnerManager.class).to(DefaultPreviewRunnerManager.class);
         expose(PreviewRunnerManager.class);
-
-        install(new FactoryModuleBuilder()
-            .implement(PreviewRunnerService.class, PreviewRunnerService.class)
-            .build(PreviewRunnerServiceFactory.class));
       }
     };
   }
@@ -90,10 +86,6 @@ protected void configure() {
         bind(DefaultPreviewRunnerManager.class).in(Scopes.SINGLETON);
         bind(PreviewRunnerManager.class).to(DefaultPreviewRunnerManager.class);
         expose(PreviewRunnerManager.class);
-
-        install(new FactoryModuleBuilder()
-            .implement(PreviewRunnerService.class, PreviewRunnerService.class)
-            .build(PreviewRunnerServiceFactory.class));
       }
     };
   }

cdap-app-fabric/src/main/java/io/cdap/cdap/app/preview/PreviewStatus.java

@@ -25,7 +25,7 @@
  */
 public class PreviewStatus {
 
   /**
    * Status for the preview
    */
   public enum Status {
@@ -37,7 +37,8 @@
     RUN_FAILED(true),
     KILLED(true),
     KILLED_BY_TIMER(true),
-    KILLED_BY_EXCEEDING_MEMORY_LIMIT(true);
+    KILLED_BY_EXCEEDING_MEMORY_LIMIT(true),
+    KILLED_BY_INSUFFICIENT_RESOURCES(true);
 
     private final boolean endState;
 
@@ -63,7 +64,7 @@
     this(status, submitTime, throwable, startTime, endTime, null);
   }
 
   public PreviewStatus(Status status, long submitTime, @Nullable BasicThrowable throwable,
       @Nullable Long startTime,
       @Nullable Long endTime, @Nullable Integer positionInWaitingQueue) {
     this.status = status;

cdap-app-fabric/src/main/java/io/cdap/cdap/app/store/preview/PreviewStore.java

@@ -13,7 +13,7 @@
  * License for the specific language governing permissions and limitations under
  * the License.
  */
 package io.cdap.cdap.app.store.preview;
 
 import com.google.gson.JsonElement;
 import io.cdap.cdap.api.preview.DataTracer;
@@ -22,6 +22,7 @@
 import io.cdap.cdap.common.ConflictException;
 import io.cdap.cdap.proto.artifact.AppRequest;
 import io.cdap.cdap.proto.id.ApplicationId;
+import io.cdap.cdap.proto.id.ProgramId;
 import io.cdap.cdap.proto.id.ProgramRunId;
 import io.cdap.cdap.proto.security.Principal;
 import java.util.List;
@@ -52,21 +53,25 @@
    */
   Map<String, List<JsonElement>> get(ApplicationId applicationId, String tracerName);
 
   /**
    * Removes the preview data stored by specified application id
    *
    * @param applicationId the id of the preview for which the data to be removed
    */
   void remove(ApplicationId applicationId);
 
   /**
    * Save the program run id associated with the preview run
    *
    * @param programRunId the program run id to save
    */
-  void setProgramId(ProgramRunId programRunId);
+  void setProgramRunId(ProgramRunId programRunId);
+
+  void setProgramId(ProgramId programId);
+
+  ProgramId getProgramId(ApplicationId applicationId);
 
   /**
    * Get the program run id associated with the preview run
    *
    * @param applicationId the preview id
@@ -75,7 +80,7 @@
   @Nullable
   ProgramRunId getProgramRunId(ApplicationId applicationId);
 
   /**
    * Set the preview status assoicated with the preview run
    *
    * @param applicationId the preview id
@@ -83,7 +88,7 @@
    */
   void setPreviewStatus(ApplicationId applicationId, PreviewStatus previewStatus);
 
   /**
    * Get the preview status assoicated with the preview run
    *
    * @param applicationId the preview id
@@ -102,7 +107,7 @@
    */
   void add(ApplicationId applicationId, AppRequest appRequest, @Nullable Principal principal);
 
   /**
    * @return list of all waiting requests in waiting state sorted by submit time
    */
   List<PreviewRequest> getAllInWaitingState();
@@ -118,7 +123,7 @@
   void setPreviewRequestPollerInfo(ApplicationId applicationId, @Nullable byte[] pollerInfo)
       throws ConflictException;
 
   /**
    * @return the poller info associated with the application if it exists, otherwise {@code null} is
    *     returned
    */

cdap-app-fabric/src/main/java/io/cdap/cdap/gateway/handlers/preview/PreviewHttpHandlerInternal.java

@@ -36,6 +36,7 @@
 /**
  * Internal {@link HttpHandler} for Preview system.
  */
+@Deprecated
 @Singleton
 @Path(Constants.Gateway.INTERNAL_API_VERSION_3 + "/previews")
 public class PreviewHttpHandlerInternal extends AbstractHttpHandler {
@@ -49,7 +50,7 @@
     this.previewManager = previewManager;
   }
 
   @POST
   @Path("/requests/pull")
   public void poll(FullHttpRequest request, HttpResponder responder) {
     byte[] pollerInfo = Bytes.toBytes(request.content().nioBuffer());

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

@@ -129,19 +129,20 @@
   private final AccessControllerInstantiator accessControllerInstantiator;
   private final AccessEnforcer accessEnforcer;
   private final AuthenticationContext authenticationContext;
   private final CConfiguration previewCConf;
   private final Configuration previewHConf;
   private final SConfiguration previewSConf;
   private final DiscoveryServiceClient discoveryServiceClient;
   private final DatasetFramework datasetFramework;
   private final TransactionSystemClient transactionSystemClient;
   private final LevelDBTableService previewLevelDBTableService;
   private final PreviewRequestQueue previewRequestQueue;
   private final PreviewStore previewStore;
   private final PreviewRunStopper previewRunStopper;
   private final MessagingService messagingService;
   private final PreviewDataCleanupService previewDataCleanupService;
   private final MetricsCollectionService metricsCollectionService;
+  private final PreviewRequestPollerService previewRequestPollerService;
   private Injector previewInjector;
   private PreviewDataSubscriberService dataSubscriberService;
   private PreviewTMSLogSubscriber logSubscriberService;
@@ -154,14 +155,15 @@
       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,
       PreviewRequestQueue previewRequestQueue, PreviewStore previewStore,
       PreviewRunStopper previewRunStopper, MessagingService messagingService,
       PreviewDataCleanupService previewDataCleanupService,
-      MetricsCollectionService metricsCollectionService) {
+      MetricsCollectionService metricsCollectionService,
+      PreviewRequestPollerService previewRequestPollerService) {
     this.authenticationContext = authenticationContext;
     this.previewCConf = previewCConf;
     this.previewHConf = previewHConf;
@@ -178,6 +180,7 @@
     this.messagingService = messagingService;
     this.previewDataCleanupService = previewDataCleanupService;
     this.metricsCollectionService = metricsCollectionService;
+    this.previewRequestPollerService = previewRequestPollerService;
   }
 
   @Override
@@ -196,10 +199,12 @@
     dataSubscriberService = previewInjector.getInstance(PreviewDataSubscriberService.class);
     dataSubscriberService.startAndWait();
     previewDataCleanupService.startAndWait();
+    previewRequestPollerService.startAndWait();
   }
 
   @Override
   protected void shutDown() throws Exception {
+    stopQuietly(previewRequestPollerService);
     stopQuietly(previewDataCleanupService);
     stopQuietly(dataSubscriberService);
     stopQuietly(logSubscriberService);
@@ -225,6 +230,8 @@
     }
 
     previewRequestQueue.add(previewRequest);
+    LOG.info("sidhdirenge - Setting program id {} and appId {}", programId, programId.getParent());
+    previewStore.setProgramId(programId);
     return previewApp;
   }
 
@@ -264,7 +271,9 @@
           status.getSubmitTime(), null, null, null));
       return;
     }
-    previewRunStopper.stop(applicationId);
+    ProgramId programId = previewStore.getProgramId(applicationId);
+    LOG.info("sidhdirenge - Got program Id {} for application {}", programId, applicationId);
+    previewRunStopper.stop(programId);
   }
 
   @Override