<commit_before>/**
 * (c) 2014 StreamSets, Inc. All rights reserved. May not
 * be copied, modified, or distributed in whole or part without
 * written consent of StreamSets, Inc.
 */
package com.streamsets.datacollector.store.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.streamsets.datacollector.config.StageConfiguration;
import com.streamsets.pipeline.api.ExecutionMode;
import com.streamsets.pipeline.api.impl.PipelineUtils;
import com.streamsets.pipeline.api.impl.Utils;
import com.streamsets.datacollector.config.DataRuleDefinition;
import com.streamsets.datacollector.config.MetricsRuleDefinition;
import com.streamsets.datacollector.config.PipelineConfiguration;
import com.streamsets.datacollector.config.RuleDefinitions;
import com.streamsets.datacollector.creation.PipelineBeanCreator;
import com.streamsets.datacollector.creation.PipelineConfigBean;
import com.streamsets.datacollector.execution.PipelineStateStore;
import com.streamsets.datacollector.execution.PipelineStatus;
import com.streamsets.datacollector.io.DataStore;
import com.streamsets.datacollector.json.ObjectMapperFactory;
import com.streamsets.datacollector.main.RuntimeInfo;
import com.streamsets.datacollector.restapi.bean.BeanHelper;
import com.streamsets.datacollector.restapi.bean.PipelineConfigurationJson;
import com.streamsets.datacollector.restapi.bean.PipelineInfoJson;
import com.streamsets.datacollector.restapi.bean.RuleDefinitionsJson;
import com.streamsets.datacollector.stagelibrary.StageLibraryTask;
import com.streamsets.datacollector.store.PipelineInfo;
import com.streamsets.datacollector.store.PipelineRevInfo;
import com.streamsets.datacollector.store.PipelineStoreException;
import com.streamsets.datacollector.store.PipelineStoreTask;
import com.streamsets.datacollector.task.AbstractTask;
import com.streamsets.datacollector.util.ContainerError;
import com.streamsets.datacollector.util.LockCache;
import com.streamsets.datacollector.util.PipelineDirectoryUtil;
import com.streamsets.datacollector.validation.Issue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.inject.Inject;

import java.io.File;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class FilePipelineStoreTask extends AbstractTask implements PipelineStoreTask {
  private static final Logger LOG = LoggerFactory.getLogger(FilePipelineStoreTask.class);
  private final LockCache<String> lockCache;
  @VisibleForTesting
  static final String REV = "0";
  public static final String INFO_FILE = "info.json";
  public static final String PIPELINE_FILE = "pipeline.json";
  public static final String UI_INFO_FILE = "uiinfo.json";
  public static final String RULES_FILE = "rules.json";

  private final StageLibraryTask stageLibrary;
  private final RuntimeInfo runtimeInfo;
  private File storeDir;
  private final ObjectMapper json;
  private final PipelineStateStore pipelineStateStore;
  private final ConcurrentMap<String, RuleDefinitions> pipelineToRuleDefinitionMap;

  @Inject
  public FilePipelineStoreTask(RuntimeInfo runtimeInfo, StageLibraryTask stageLibrary,
    PipelineStateStore pipelineStateStore, LockCache<String> lockCache) {
    super("filePipelineStore");
    this.stageLibrary = stageLibrary;
    this.runtimeInfo = runtimeInfo;
    json = ObjectMapperFactory.get();
    pipelineToRuleDefinitionMap = new ConcurrentHashMap<>();
    this.pipelineStateStore = pipelineStateStore;
    this.lockCache = lockCache;
  }

  @VisibleForTesting
  File getStoreDir() {
    return storeDir;
  }

  @Override
  public void initTask() {
    storeDir = new File(runtimeInfo.getDataDir(), PipelineDirectoryUtil.PIPELINE_INFO_BASE_DIR);
    if (!storeDir.exists()) {
      if (!storeDir.mkdirs()) {
        throw new RuntimeException(Utils.format("Could not create directory '{}'", storeDir.getAbsolutePath()));
      }
    }
    if (pipelineStateStore != null) {
      pipelineStateStore.init();
    }
  }

  @Override
  public void stopTask() {
    if (pipelineStateStore != null) {
      pipelineStateStore.destroy();
    }
  }

  public File getPipelineDir(String name) {
    return new File(storeDir, PipelineUtils.escapedPipelineName(name));
  }

  @VisibleForTesting
  File getInfoFile(String name) {
    return new File(getPipelineDir(name), INFO_FILE);
  }

  private File getPipelineFile(String name) {
    return new File(getPipelineDir(name), PIPELINE_FILE);
  }

  private File getPipelineUiInfoFile(String name) {
    return new File(getPipelineDir(name), UI_INFO_FILE);
  }

  private File getRulesFile(String name) {
    return new File(getPipelineDir(name), RULES_FILE);
  }

  @Override
  public boolean hasPipeline(String name) {
    return getPipelineDir(name).exists();
  }

  @Override
  public PipelineConfiguration create(String user, String name, String description) throws PipelineStoreException {
    synchronized (lockCache.getLock(name)) {
      if (hasPipeline(name)) {
        throw new PipelineStoreException(ContainerError.CONTAINER_0201, name);
      }
      if (!getPipelineDir(name).mkdir()) {
        throw new PipelineStoreException(ContainerError.CONTAINER_0202, name, Utils.format("'{}' mkdir failed",
          getPipelineDir(name)));
      }

      Date date = new Date();
      UUID uuid = UUID.randomUUID();
      PipelineInfo info = new PipelineInfo(name, description, date, date, user, user, REV, uuid, false);
      PipelineConfiguration pipeline =
        new PipelineConfiguration(SCHEMA_VERSION, PipelineConfigBean.VERSION, uuid, description, stageLibrary
          .getPipeline().getPipelineDefaultConfigs(), Collections.EMPTY_MAP, Collections.EMPTY_LIST, null);

      try {
        json.writeValue(getInfoFile(name), BeanHelper.wrapPipelineInfo(info));
        json.writeValue(getPipelineFile(name), BeanHelper.wrapPipelineConfiguration(pipeline));
      } catch (Exception ex) {
        throw new PipelineStoreException(ContainerError.CONTAINER_0202, name, ex.toString(), ex);
      }
      pipeline.setPipelineInfo(info);
      if (pipelineStateStore != null) {
        pipelineStateStore.saveState(user, name, REV, PipelineStatus.EDITED, "Pipeline edited", null,
          ExecutionMode.STANDALONE, null, 0, 0);
      }
      return pipeline;
    }
  }

  private boolean cleanUp(String name) throws PipelineStoreException {
    boolean deleted = PipelineDirectoryUtil.deleteAll(getPipelineDir(name));
    deleted &= PipelineDirectoryUtil.deletePipelineDir(runtimeInfo, name);
    return deleted;
  }

  @Override
  public void delete(String name) throws PipelineStoreException {
    synchronized (lockCache.getLock(name)) {
      if (!hasPipeline(name)) {
        throw new PipelineStoreException(ContainerError.CONTAINER_0200, name);
      }
      if (pipelineStateStore != null) {
        // For now, passing rev 0 - make delete take tag/rev as a parameter
        PipelineStatus pipelineStatus = pipelineStateStore.getState(name, REV).getStatus();
        if (pipelineStatus.isActive()) {
          throw new PipelineStoreException(ContainerError.CONTAINER_0208, pipelineStatus);
        }
        if (!cleanUp(name)) {
          throw new PipelineStoreException(ContainerError.CONTAINER_0203, name);
        }<commit_msg>SDC-1552. Deleting pipeline doesn't delete state history, offset in runInfo<commit_after>/**
 * (c) 2014 StreamSets, Inc. All rights reserved. May not
 * be copied, modified, or distributed in whole or part without
 * written consent of StreamSets, Inc.
 */
package com.streamsets.datacollector.store.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.streamsets.datacollector.config.StageConfiguration;
import com.streamsets.datacollector.util.LogUtil;
import com.streamsets.pipeline.api.ExecutionMode;
import com.streamsets.pipeline.api.impl.PipelineUtils;
import com.streamsets.pipeline.api.impl.Utils;
import com.streamsets.datacollector.config.DataRuleDefinition;
import com.streamsets.datacollector.config.MetricsRuleDefinition;
import com.streamsets.datacollector.config.PipelineConfiguration;
import com.streamsets.datacollector.config.RuleDefinitions;
import com.streamsets.datacollector.creation.PipelineBeanCreator;
import com.streamsets.datacollector.creation.PipelineConfigBean;
import com.streamsets.datacollector.execution.PipelineStateStore;
import com.streamsets.datacollector.execution.PipelineStatus;
import com.streamsets.datacollector.io.DataStore;
import com.streamsets.datacollector.json.ObjectMapperFactory;
import com.streamsets.datacollector.main.RuntimeInfo;
import com.streamsets.datacollector.restapi.bean.BeanHelper;
import com.streamsets.datacollector.restapi.bean.PipelineConfigurationJson;
import com.streamsets.datacollector.restapi.bean.PipelineInfoJson;
import com.streamsets.datacollector.restapi.bean.RuleDefinitionsJson;
import com.streamsets.datacollector.stagelibrary.StageLibraryTask;
import com.streamsets.datacollector.store.PipelineInfo;
import com.streamsets.datacollector.store.PipelineRevInfo;
import com.streamsets.datacollector.store.PipelineStoreException;
import com.streamsets.datacollector.store.PipelineStoreTask;
import com.streamsets.datacollector.task.AbstractTask;
import com.streamsets.datacollector.util.ContainerError;
import com.streamsets.datacollector.util.LockCache;
import com.streamsets.datacollector.util.PipelineDirectoryUtil;
import com.streamsets.datacollector.validation.Issue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.inject.Inject;

import java.io.File;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class FilePipelineStoreTask extends AbstractTask implements PipelineStoreTask {
  private static final Logger LOG = LoggerFactory.getLogger(FilePipelineStoreTask.class);
  private final LockCache<String> lockCache;
  @VisibleForTesting
  static final String REV = "0";
  public static final String INFO_FILE = "info.json";
  public static final String PIPELINE_FILE = "pipeline.json";
  public static final String UI_INFO_FILE = "uiinfo.json";
  public static final String RULES_FILE = "rules.json";
  public static final String STATE = "state";

  private final StageLibraryTask stageLibrary;
  private final RuntimeInfo runtimeInfo;
  private File storeDir;
  private final ObjectMapper json;
  private final PipelineStateStore pipelineStateStore;
  private final ConcurrentMap<String, RuleDefinitions> pipelineToRuleDefinitionMap;

  @Inject
  public FilePipelineStoreTask(RuntimeInfo runtimeInfo, StageLibraryTask stageLibrary,
    PipelineStateStore pipelineStateStore, LockCache<String> lockCache) {
    super("filePipelineStore");
    this.stageLibrary = stageLibrary;
    this.runtimeInfo = runtimeInfo;
    json = ObjectMapperFactory.get();
    pipelineToRuleDefinitionMap = new ConcurrentHashMap<>();
    this.pipelineStateStore = pipelineStateStore;
    this.lockCache = lockCache;
  }

  @VisibleForTesting
  File getStoreDir() {
    return storeDir;
  }

  @Override
  public void initTask() {
    storeDir = new File(runtimeInfo.getDataDir(), PipelineDirectoryUtil.PIPELINE_INFO_BASE_DIR);
    if (!storeDir.exists()) {
      if (!storeDir.mkdirs()) {
        throw new RuntimeException(Utils.format("Could not create directory '{}'", storeDir.getAbsolutePath()));
      }
    }
    if (pipelineStateStore != null) {
      pipelineStateStore.init();
    }
  }

  @Override
  public void stopTask() {
    if (pipelineStateStore != null) {
      pipelineStateStore.destroy();
    }
  }

  public File getPipelineDir(String name) {
    return new File(storeDir, PipelineUtils.escapedPipelineName(name));
  }

  @VisibleForTesting
  File getInfoFile(String name) {
    return new File(getPipelineDir(name), INFO_FILE);
  }

  private File getPipelineFile(String name) {
    return new File(getPipelineDir(name), PIPELINE_FILE);
  }

  private File getPipelineUiInfoFile(String name) {
    return new File(getPipelineDir(name), UI_INFO_FILE);
  }

  private File getRulesFile(String name) {
    return new File(getPipelineDir(name), RULES_FILE);
  }

  @Override
  public boolean hasPipeline(String name) {
    return getPipelineDir(name).exists();
  }

  @Override
  public PipelineConfiguration create(String user, String name, String description) throws PipelineStoreException {
    synchronized (lockCache.getLock(name)) {
      if (hasPipeline(name)) {
        throw new PipelineStoreException(ContainerError.CONTAINER_0201, name);
      }
      if (!getPipelineDir(name).mkdir()) {
        throw new PipelineStoreException(ContainerError.CONTAINER_0202, name, Utils.format("'{}' mkdir failed",
          getPipelineDir(name)));
      }

      Date date = new Date();
      UUID uuid = UUID.randomUUID();
      PipelineInfo info = new PipelineInfo(name, description, date, date, user, user, REV, uuid, false);
      PipelineConfiguration pipeline =
        new PipelineConfiguration(SCHEMA_VERSION, PipelineConfigBean.VERSION, uuid, description, stageLibrary
          .getPipeline().getPipelineDefaultConfigs(), Collections.EMPTY_MAP, Collections.EMPTY_LIST, null);

      try {
        json.writeValue(getInfoFile(name), BeanHelper.wrapPipelineInfo(info));
        json.writeValue(getPipelineFile(name), BeanHelper.wrapPipelineConfiguration(pipeline));
      } catch (Exception ex) {
        throw new PipelineStoreException(ContainerError.CONTAINER_0202, name, ex.toString(), ex);
      }
      pipeline.setPipelineInfo(info);
      if (pipelineStateStore != null) {
        pipelineStateStore.saveState(user, name, REV, PipelineStatus.EDITED, "Pipeline edited", null,
          ExecutionMode.STANDALONE, null, 0, 0);
      }
      return pipeline;
    }
  }

  private boolean cleanUp(String name) throws PipelineStoreException {
    boolean deleted = PipelineDirectoryUtil.deleteAll(getPipelineDir(name));
    deleted &= PipelineDirectoryUtil.deletePipelineDir(runtimeInfo, name);
    if(deleted) {
      LogUtil.resetRollingFileAppender(name, "0", STATE);
    }
    return deleted;
  }

  @Override
  public void delete(String name) throws PipelineStoreException {
    synchronized (lockCache.getLock(name)) {
      if (!hasPipeline(name)) {
        throw new PipelineStoreException(ContainerError.CONTAINER_0200, name);
      }
      if (pipelineStateStore != null) {
        // For now, passing rev 0 - make delete take tag/rev as a parameter
        PipelineStatus pipelineStatus = pipelineStateStore.getState(name, REV).getStatus();
        if (pipelineStatus.isActive()) {
          throw new PipelineStoreException(ContainerError.CONTAINER_0208, pipelineStatus);
        }
        if (!cleanUp(name)) {
          throw new PipelineStoreException(ContainerError.CONTAINER_0203, name);
        }
{'fraction_non_alphanumeric': 0.09272844272844273, 'fraction_numerical': 0.004697554697554698, 'mean_word_length': 4.851280120481928, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3233146', 'n_tokens_mistral': 4470, 'n_tokens_neox': 4459, 'n_words': 929}