content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you 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
   * <p>
   * http://www.apache.org/licenses/LICENSE-2.0
   * <p>
   * 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 org.apache.ambari.server.view;
  
  import static com.google.common.base.Preconditions.checkArgument;
  import static java.lang.Thread.sleep;
  
  import java.io.File;
  import java.io.IOException;
  import java.nio.file.Path;
  import java.nio.file.Paths;
  import java.nio.file.StandardWatchEventKinds;
  import java.nio.file.WatchEvent;
  import java.nio.file.WatchKey;
  import java.nio.file.WatchService;
  import java.util.Collections;
  import java.util.List;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.zip.ZipFile;
  
  import javax.annotation.Nullable;
  
  import org.apache.ambari.server.configuration.Configuration;
  import org.apache.ambari.server.utils.Closeables;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.google.common.base.Function;
  import com.google.common.collect.Lists;
  import com.google.inject.Inject;
  import com.google.inject.Singleton;
  
  @Singleton
  public class ViewDirectoryWatcher implements DirectoryWatcher {
  
    public static final int FIXED_FILE_COUNTER = 30;
    public static final int FILE_CHECK_INTERVAL_MILLIS = 200;
    // Global configuration
    @Inject
    Configuration configuration;
  
    // View Registry
    @Inject
    ViewRegistry viewRegistry;
  
    private WatchService watchService;
  
    // Executor service on which the watcher will run
    private ExecutorService executorService = Executors.newSingleThreadExecutor();
  
    private Future<?> watchTask;
  
    private static final Logger LOG = LoggerFactory.getLogger(ViewDirectoryWatcher.class);
  
    // Callbacks to hook into file processing
    private List<Function<Path, Boolean>> hooks = Lists.newArrayList(Collections.singleton(loggingHook()));
  
    public void addHook(Function<Path, Boolean> hook) {
      hooks.add(hook);
    }
  
    private Function<Path, Boolean> loggingHook() {
      return new Function<Path, Boolean>() {
        @Nullable
        @Override
        public Boolean apply(@Nullable Path path) {
          LOG.info("Finished processing the view definition for" + path);
          return true;
        }
      };
    }
  
    @Override
    public void start() {
  
      try {
        Path path = buildWatchService();
        Runnable task = startWatching(path);
        watchTask = executorService.submit(task);
      } catch (Exception e) {
        LOG.error("There were errors in starting the view directory watcher. This task will not run", e);
      }
    }
  
  
    @SuppressWarnings("unchecked")
    private static <T> WatchEvent<T> cast(WatchEvent<?> event) {
      return (WatchEvent<T>) event;
    }
  
    private Runnable startWatching(final Path path) {
      return new Runnable() {
        @Override
        public void run() {
          try {
            while (true) {
              // wait for key , park the thread meanwhile
              WatchKey key = watchService.take();
              LOG.info("Watcher Key was signalled");
              for (WatchEvent<?> event : key.pollEvents()) {
                LOG.info("Watcher recieved poll event");
                WatchEvent<Path> ev = cast(event);
                Path resolvedPath = path.resolve(ev.context());
                LOG.info(String.format("Event %s: %s\n", ev.kind(), resolvedPath));
                if (!canBlockTillFileAvailable(resolvedPath)) {
                  LOG.info("Watcher detected that the file was either empty or corrupt");
                  continue;
                }
                if (!verify(resolvedPath)) {
                  LOG.info("The uploaded file was 1> Empty 2> Not a regular file or 3> Not a valid Jar archive file");
                  continue;
                }
                try {
                  LOG.info("Starting view extraction");
                  viewRegistry.readViewArchive(resolvedPath);
                  // fire registered hooks
                  for (Function<Path, Boolean> hook : hooks) {
                    hook.apply(resolvedPath);
                  }
                } catch (Exception e) {
                  LOG.error("Cannot read the view archive, offending file: " + resolvedPath, e);
                }
  
              }
  
              // reset key
              if (!key.reset()) {
                //watch key is invalid, break out
                LOG.error("The watch key could not be reset, Directory watcher will not run anymore");
                break;
              }
  
  
            }
          } catch (InterruptedException x) {
            LOG.info("Cancelling the directory watcher", x);
            return;
          }
  
        }
      };
    }
  
  
    /**
     * Routine to make the file watcher block the thread till the file is completely copied
     * Check the length of the file continuously till there are 20 consecutive intervals when
     * the file length does not change
     * FILE_CHECK_INTERVAL_MILLIS defines the check interval both for detecting empty files
     * and subsequent checks to detect if a file has finished copying
     *
     * The process which copies the jar into the views dir is external and we dont really
     * know when it would finish, this is also highly OS and FS dependent. The following routine
     * introduces a heuristic to detect when a file has finished copying by looking at subsequent
     * lengths of the file which was detected as being created
     *
     * This would block for ~ 7 seconds in most cases
     *
     *
     * @param resolvedPath
     * @return false if the file check failed, true otherwise
     */
    private boolean canBlockTillFileAvailable(Path resolvedPath) throws InterruptedException {
      long oldLength;
      long newSize;
  
      long emptyCheck = 0;
      int fixed = 0;
      // get the underlying file
      File file = resolvedPath.toAbsolutePath().toFile();
  
      // empty file check
      while (file.length() == 0 && emptyCheck < 5) {
        sleep(FILE_CHECK_INTERVAL_MILLIS);
        emptyCheck++;
      }
      // The file seems to be empty
      if (emptyCheck == 5)
        return false;
  
      // check the file size
      oldLength = file.length();
  
      // Check if file copy is done
      while (true) {
        LOG.info("Waiting for file to be completely copied");
        sleep(FILE_CHECK_INTERVAL_MILLIS);
        newSize = file.length();
        if (newSize > oldLength) {
          oldLength = newSize;
          continue;
        } else if (oldLength == newSize) {
          fixed++;
        } else {
          // this can never happen,
          return false;
        }
        if (fixed > FIXED_FILE_COUNTER) {
          LOG.info("File " + resolvedPath + " has finished copying");
          return true;
        }
      }
  
  
    }
  
    /**
     * Sanity check to validate if the detected path is a valid archive file
     * is not a directory, also check that the file is not empty
     *
     * @param resolvedPath
     * @return
     */
    private boolean verify(Path resolvedPath) {
      ZipFile zipFile = null;
      try {
        File file = resolvedPath.toAbsolutePath().toFile();
        checkArgument(!file.isDirectory());
        checkArgument(file.length() > 0);
        zipFile = new ZipFile(file);
      } catch (Exception e) {
        LOG.info("Verification failed ", e);
        return false;
      } finally {
        Closeables.closeSilently(zipFile);
      }
      return true;
    }
  
    private Path buildWatchService() throws IOException {
      // Get the directory for view Archives
      //Attach a file watcher at this directory, Extracted work directory will be ignored
      File viewsDir = configuration.getViewsDir();
      Path path = Paths.get(viewsDir.getAbsolutePath());
  
      watchService = path.getFileSystem().newWatchService();
      //Watch vews directory for creation events
      path.register(watchService, StandardWatchEventKinds.ENTRY_CREATE);
      return path;
  
    }
  
    @Override
    public boolean isRunning() {
      if (watchTask != null)
        return !(watchTask.isDone());
      return false;
    }
  
    @Override
    public void stop() {
      watchTask.cancel(true);
    }
  
  }
  