src/main/java/org/apache/hadoop/hdfs/server/namenode/analytics/web/NamenodeAnalyticsMethods.java
@GET
  @Path("/submitOperation")
  @Produces({MediaType.TEXT_PLAIN})
  public Response submitOperation() {
    final NameNodeLoader nnLoader = (NameNodeLoader) context.getAttribute(NNA_NN_LOADER);
    final SecurityContext securityContext =
        (SecurityContext) context.getAttribute(NNA_SECURITY_CONTEXT);
    final ReentrantReadWriteLock queryLock =
        (ReentrantReadWriteLock) context.getAttribute(NNA_QUERY_LOCK);
    final ApplicationConfiguration conf =
        (ApplicationConfiguration) context.getAttribute(NNA_APP_CONF);
    @SuppressWarnings("unchecked")
    final Map<String, BaseOperation> runningOperations =
        (Map<String, BaseOperation>) context.getAttribute(NNA_RUNNING_OPERATIONS);
    final ExecutorService operationService =
        (ExecutorService) context.getAttribute(NNA_OPERATION_SERVICE);
    try {

      if (!nnLoader.isInit()) {
        response.setHeader("Content-Type", "application/json");
        String message = "Not accepting operations yet.";
        PrintWriter writer = response.getWriter();
        writer.write(message);
        writer.flush();
        writer.close();
        return Response.ok().build();
      }

      queryLock.writeLock().lock();
      try {
        final String fullFilterStr = request.getParameter("filters");
        final String find = request.getParameter("find");
        final String[] filters = Helper.parseFilters(fullFilterStr);
        final String[] filterOps = Helper.parseFilterOps(fullFilterStr);
        final String set = request.getParameter("set");
        final String operation = request.getParameter("operation");
        String sleepStr = request.getParameter("sleep");
        Integer sleep;
        if (sleepStr == null) {
          sleep = Constants.DEFAULT_DELETE_SLEEP_MS;
        } else {
          sleep = Integer.parseInt(sleepStr);
        }

        if (operation == null || operation.isEmpty()) {
          throw new IllegalArgumentException("No operation defined. Please check /operations.");
        }
        QueryChecker.isValidQuery(set, filters, null, null, filterOps, find);

        Collection<INode> filteredINodes =
            Helper.performFilters(nnLoader, set, filters, filterOps, find);
        if (filteredINodes.size() == 0) {
          LOG.info("Skipping operation request because it resulted in empty INode set.");
          throw new IOException(
              "Skipping operation request because it resulted in empty INode set.");
        }

        FileSystem fs = nnLoader.getFileSystem();
        String[] operationSplits = operation.split(":");
        String logBaseDir = conf.getBaseDir();
        BaseOperation operationObj;
        switch (operationSplits[0]) {
          case "delete":
            operationObj =
                new Delete(
                    filteredINodes,
                    request.getQueryString(),
                    securityContext.getUserName(),
                    logBaseDir,
                    fs);
            break;
          case "setReplication":
            short newReplFactor = Short.parseShort(operationSplits[1]);
            operationObj =
                new SetReplication(
                    filteredINodes,
                    request.getQueryString(),
                    securityContext.getUserName(),
                    logBaseDir,
                    fs,
                    newReplFactor);
            break;
          case "setStoragePolicy":
            operationObj =
                new SetStoragePolicy(
                    filteredINodes,
                    request.getQueryString(),
                    securityContext.getUserName(),
                    logBaseDir,
                    fs,
                    operationSplits[1]);
            break;
          default:
            throw new IllegalArgumentException(
                "Unknown operation:" + operationSplits[0] + ". Please check /operations.");
        }
        runningOperations.put(operationObj.identity(), operationObj);

        final int finalSleep = sleep;
        Callable<Boolean> callable =
            () -> {
              try {
                operationObj.initialize();
                while (operationObj.hasNext()) {
                  try {
                    if (finalSleep >= 100) {
                      Thread.sleep(finalSleep);
                    }
                  } catch (InterruptedException ignored) {
                    LOG.debug("Operation sleep interrupted due to: ", ignored);
                  }
                  operationObj.performOp();
                }
                operationObj.close();
                return Boolean.TRUE;
              } catch (IllegalStateException e) {
                operationObj.abort();
                LOG.info("Aborted operation due to: ", e);
                return Boolean.FALSE;
              } finally {
                runningOperations.remove(operationObj.identity());
              }
            };
        Future<Boolean> submittedOperation = operationService.submit(callable);
        operationObj.linkFuture(submittedOperation);

        PrintWriter writer = response.getWriter();
        writer.write(operationObj.identity());
        writer.flush();
        writer.close();
        return Response.ok().build();
      } finally {
        queryLock.writeLock().unlock();
      }
    } catch (RuntimeException rtex) {
      return handleException(rtex);
    } catch (Exception ex) {
      return handleException(ex);
    } finally {
      after();
    }
  }
/**
   * SUBMITOPERATION is a GET only call that only WRITER users can access. It take parameters like
   * /filter and schedules operations to perform.
   */
SUBMITOPERATION is a GET only call that only WRITER users can access. It take parameters like
/filter and schedules operations to perform.
SUBMITOPERATION is a GET only call that only WRITER users can access . It take parameters like / filter and schedules operations to perform .
@GET
  @Path("/submitOperation")
  @Produces({MediaType.TEXT_PLAIN})
  public Response submitOperation() {
    final NameNodeLoader nnLoader = (NameNodeLoader) context.getAttribute(NNA_NN_LOADER);
    final SecurityContext securityContext =
        (SecurityContext) context.getAttribute(NNA_SECURITY_CONTEXT);
    final ReentrantReadWriteLock queryLock =
        (ReentrantReadWriteLock) context.getAttribute(NNA_QUERY_LOCK);
    final ApplicationConfiguration conf =
        (ApplicationConfiguration) context.getAttribute(NNA_APP_CONF);
    @SuppressWarnings("unchecked")
    final Map<String, BaseOperation> runningOperations =
        (Map<String, BaseOperation>) context.getAttribute(NNA_RUNNING_OPERATIONS);
    final ExecutorService operationService =
        (ExecutorService) context.getAttribute(NNA_OPERATION_SERVICE);
    try {
      if (!nnLoader.isInit()) {
        response.setHeader("Content-Type", "application/json");
        String message = "Not accepting operations yet.";
        PrintWriter writer = response.getWriter();
        writer.write(message);
        writer.flush();
        writer.close();
        return Response.ok().build();
      }
      queryLock.writeLock().lock();
      try {
        final String fullFilterStr = request.getParameter("filters");
        final String find = request.getParameter("find");
        final String[] filters = Helper.parseFilters(fullFilterStr);
        final String[] filterOps = Helper.parseFilterOps(fullFilterStr);
        final String set = request.getParameter("set");
        final String operation = request.getParameter("operation");
        String sleepStr = request.getParameter("sleep");
        Integer sleep;
        if (sleepStr == null) {
          sleep = Constants.DEFAULT_DELETE_SLEEP_MS;
        } else {
          sleep = Integer.parseInt(sleepStr);
        }
        if (operation == null || operation.isEmpty()) {
          throw new IllegalArgumentException("No operation defined. Please check /operations.");
        }
        QueryChecker.isValidQuery(set, filters, null, null, filterOps, find);
        Collection<INode> filteredINodes =
            Helper.performFilters(nnLoader, set, filters, filterOps, find);
        if (filteredINodes.size() == 0) {
          LOG.info("Skipping operation request because it resulted in empty INode set.");
          throw new IOException(
              "Skipping operation request because it resulted in empty INode set.");
        }
        FileSystem fs = nnLoader.getFileSystem();
        String[] operationSplits = operation.split(":");
        String logBaseDir = conf.getBaseDir();
        BaseOperation operationObj;
        switch (operationSplits[0]) {
          case "delete":
            operationObj =
                new Delete(
                    filteredINodes,
                    request.getQueryString(),
                    securityContext.getUserName(),
                    logBaseDir,
                    fs);
            break;
          case "setReplication":
            short newReplFactor = Short.parseShort(operationSplits[1]);
            operationObj =
                new SetReplication(
                    filteredINodes,
                    request.getQueryString(),
                    securityContext.getUserName(),
                    logBaseDir,
                    fs,
                    newReplFactor);
            break;
          case "setStoragePolicy":
            operationObj =
                new SetStoragePolicy(
                    filteredINodes,
                    request.getQueryString(),
                    securityContext.getUserName(),
                    logBaseDir,
                    fs,
                    operationSplits[1]);
            break;
          default:
            throw new IllegalArgumentException(
                "Unknown operation:" + operationSplits[0] + ". Please check /operations.");
        }
        runningOperations.put(operationObj.identity(), operationObj);
        final int finalSleep = sleep;
        Callable<Boolean> callable =
            () -> {
              try {
                operationObj.initialize();
                while (operationObj.hasNext()) {
                  try {
                    if (finalSleep >= 100) {
                      Thread.sleep(finalSleep);
                    }
                  } catch (InterruptedException ignored) {
                    LOG.debug("Operation sleep interrupted due to: ", ignored);
                  }
                  operationObj.performOp();
                }
                operationObj.close();
                return Boolean.TRUE;
              } catch (IllegalStateException e) {
                operationObj.abort();
                LOG.info("Aborted operation due to: ", e);
                return Boolean.FALSE;
              } finally {
                runningOperations.remove(operationObj.identity());
              }
            };
        Future<Boolean> submittedOperation = operationService.submit(callable);
        operationObj.linkFuture(submittedOperation);
        PrintWriter writer = response.getWriter();
        writer.write(operationObj.identity());
        writer.flush();
        writer.close();
        return Response.ok().build();
      } finally {
        queryLock.writeLock().unlock();
      }
    } catch (RuntimeException rtex) {
      return handleException(rtex);
    } catch (Exception ex) {
      return handleException(ex);
    } finally {
      after();
    }
  }
@ GET @ Path ( "/submitOperation" ) @ Produces ( { MediaType . TEXT_PLAIN } ) public Response submitOperation ( ) { final NameNodeLoader nnLoader = ( NameNodeLoader ) context . getAttribute ( NNA_NN_LOADER ) ; final SecurityContext securityContext = ( SecurityContext ) context . getAttribute ( NNA_SECURITY_CONTEXT ) ; final ReentrantReadWriteLock queryLock = ( ReentrantReadWriteLock ) context . getAttribute ( NNA_QUERY_LOCK ) ; final ApplicationConfiguration conf = ( ApplicationConfiguration ) context . getAttribute ( NNA_APP_CONF ) ; @ SuppressWarnings ( "unchecked" ) final Map < String , BaseOperation > runningOperations = ( Map < String , BaseOperation > ) context . getAttribute ( NNA_RUNNING_OPERATIONS ) ; final ExecutorService operationService = ( ExecutorService ) context . getAttribute ( NNA_OPERATION_SERVICE ) ; try { if ( ! nnLoader . isInit ( ) ) { response . setHeader ( "Content-Type" , "application/json" ) ; String message = "Not accepting operations yet." ; PrintWriter writer = response . getWriter ( ) ; writer . write ( message ) ; writer . flush ( ) ; writer . close ( ) ; return Response . ok ( ) . build ( ) ; } queryLock . writeLock ( ) . lock ( ) ; try { final String fullFilterStr = request . getParameter ( "filters" ) ; final String find = request . getParameter ( "find" ) ; final String [ ] filters = Helper . parseFilters ( fullFilterStr ) ; final String [ ] filterOps = Helper . parseFilterOps ( fullFilterStr ) ; final String set = request . getParameter ( "set" ) ; final String operation = request . getParameter ( "operation" ) ; String sleepStr = request . getParameter ( "sleep" ) ; Integer sleep ; if ( sleepStr == null ) { sleep = Constants . DEFAULT_DELETE_SLEEP_MS ; } else { sleep = Integer . parseInt ( sleepStr ) ; } if ( operation == null || operation . isEmpty ( ) ) { throw new IllegalArgumentException ( "No operation defined. Please check /operations." ) ; } QueryChecker . isValidQuery ( set , filters , null , null , filterOps , find ) ; Collection < INode > filteredINodes = Helper . performFilters ( nnLoader , set , filters , filterOps , find ) ; if ( filteredINodes . size ( ) == 0 ) { LOG . info ( "Skipping operation request because it resulted in empty INode set." ) ; throw new IOException ( "Skipping operation request because it resulted in empty INode set." ) ; } FileSystem fs = nnLoader . getFileSystem ( ) ; String [ ] operationSplits = operation . split ( ":" ) ; String logBaseDir = conf . getBaseDir ( ) ; BaseOperation operationObj ; switch ( operationSplits [ 0 ] ) { case "delete" : operationObj = new Delete ( filteredINodes , request . getQueryString ( ) , securityContext . getUserName ( ) , logBaseDir , fs ) ; break ; case "setReplication" : short newReplFactor = Short . parseShort ( operationSplits [ 1 ] ) ; operationObj = new SetReplication ( filteredINodes , request . getQueryString ( ) , securityContext . getUserName ( ) , logBaseDir , fs , newReplFactor ) ; break ; case "setStoragePolicy" : operationObj = new SetStoragePolicy ( filteredINodes , request . getQueryString ( ) , securityContext . getUserName ( ) , logBaseDir , fs , operationSplits [ 1 ] ) ; break ; default : throw new IllegalArgumentException ( "Unknown operation:" + operationSplits [ 0 ] + ". Please check /operations." ) ; } runningOperations . put ( operationObj . identity ( ) , operationObj ) ; final int finalSleep = sleep ; Callable < Boolean > callable = ( ) -> { try { operationObj . initialize ( ) ; while ( operationObj . hasNext ( ) ) { try { if ( finalSleep >= 100 ) { Thread . sleep ( finalSleep ) ; } } catch ( InterruptedException ignored ) { LOG . debug ( "Operation sleep interrupted due to: " , ignored ) ; } operationObj . performOp ( ) ; } operationObj . close ( ) ; return Boolean . TRUE ; } catch ( IllegalStateException e ) { operationObj . abort ( ) ; LOG . info ( "Aborted operation due to: " , e ) ; return Boolean . FALSE ; } finally { runningOperations . remove ( operationObj . identity ( ) ) ; } } ; Future < Boolean > submittedOperation = operationService . submit ( callable ) ; operationObj . linkFuture ( submittedOperation ) ; PrintWriter writer = response . getWriter ( ) ; writer . write ( operationObj . identity ( ) ) ; writer . flush ( ) ; writer . close ( ) ; return Response . ok ( ) . build ( ) ; } finally { queryLock . writeLock ( ) . unlock ( ) ; } } catch ( RuntimeException rtex ) { return handleException ( rtex ) ; } catch ( Exception ex ) { return handleException ( ex ) ; } finally { after ( ) ; } }
SUBMITOPERATION is a GET only call that only WRITER users can access.
SUBMITOPERATION is a GET only call that only WRITER users can access .