<commit_before>/*
 * Copyright 2008-2013 LinkedIn, 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 voldemort.store.readonly.mr.azkaban;

import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.apache.avro.Schema;
import org.apache.avro.mapred.AvroInputFormat;
import org.apache.commons.lang.Validate;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.log4j.Logger;

import voldemort.VoldemortException;
import voldemort.client.BootstrapFailureException;
import voldemort.client.ClientConfig;
import voldemort.client.protocol.admin.AdminClient;
import voldemort.client.protocol.admin.AdminClientConfig;
import voldemort.client.protocol.pb.VAdminProto;
import voldemort.cluster.Cluster;
import voldemort.cluster.Node;
import voldemort.serialization.DefaultSerializerFactory;
import voldemort.server.VoldemortConfig;
import voldemort.store.StoreDefinition;
import voldemort.store.UnreachableStoreException;
import voldemort.store.readonly.checksum.CheckSum;
import voldemort.store.readonly.checksum.CheckSum.CheckSumType;
import voldemort.store.readonly.disk.HadoopStoreWriter;
import voldemort.store.readonly.disk.KeyValueWriter;
import voldemort.store.readonly.hooks.BuildAndPushHook;
import voldemort.store.readonly.hooks.BuildAndPushStatus;
import voldemort.store.readonly.mr.AbstractStoreBuilderConfigurable;
import voldemort.store.readonly.mr.AvroStoreBuilderMapper;
import voldemort.store.readonly.mr.HadoopStoreBuilder;
import voldemort.store.readonly.mr.JsonStoreBuilderMapper;
import voldemort.store.readonly.mr.serialization.JsonSequenceFileInputFormat;
import voldemort.store.readonly.mr.utils.AvroUtils;
import voldemort.store.readonly.mr.utils.HadoopUtils;
import voldemort.store.readonly.mr.utils.JsonSchema;
import voldemort.store.readonly.mr.utils.VoldemortUtils;
import voldemort.store.readonly.swapper.DeleteAllFailedFetchStrategy;
import voldemort.store.readonly.swapper.DisableStoreOnFailedNodeFailedFetchStrategy;
import voldemort.store.readonly.swapper.FailedFetchStrategy;
import voldemort.store.readonly.swapper.RecoverableFailedFetchException;
import voldemort.utils.ExceptionUtils;
import voldemort.utils.Props;
import voldemort.utils.ReflectUtils;
import voldemort.utils.Utils;
import azkaban.jobExecutor.AbstractJob;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.protobuf.UninitializedMessageException;

public class VoldemortBuildAndPushJob extends AbstractJob {

    private final Logger log;

    // CONFIG NAME CONSTANTS

    // build.required
    public final static String BUILD_INPUT_PATH = "build.input.path";
    public final static String BUILD_OUTPUT_DIR = "build.output.dir";
    // build.optional
    public final static String BUILD_TEMP_DIR = "build.temp.dir";
    public final static String BUILD_REPLICATION_FACTOR = "build.replication.factor";
    public final static String BUILD_COMPRESS_VALUE = "build.compress.value";
    public final static String BUILD_CHUNK_SIZE = "build.chunk.size";
    public final static String BUILD_OUTPUT_KEEP = "build.output.keep";
    public final static String BUILD_TYPE_AVRO = "build.type.avro";
    public final static String BUILD_REQUIRED_READS = "build.required.reads";
    public final static String BUILD_REQUIRED_WRITES = "build.required.writes";
    public final static String BUILD_FORCE_SCHEMA_KEY = "build.force.schema.key";
    public final static String BUILD_FORCE_SCHEMA_VALUE = "build.force.schema.value";
    public final static String BUILD_PREFERRED_READS = "build.preferred.reads";
    public final static String BUILD_PREFERRED_WRITES = "build.preferred.writes";
    public final static String BUILD_PRIMARY_REPLICAS_ONLY = "build.primary.replicas.only";
    // push.required
    public final static String PUSH_STORE_NAME = "push.store.name";
    public final static String PUSH_CLUSTER = "push.cluster";
    public final static String PUSH_STORE_OWNERS = "push.store.owners";
    public final static String PUSH_STORE_DESCRIPTION = "push.store.description";
    public final static String ENABLE_STORE_CREATION = "enable.store.creation";
    // push.optional
    public final static String PUSH_HTTP_TIMEOUT_SECONDS = "push.http.timeout.seconds";
    public final static String PUSH_VERSION = "push.version";
    public final static String PUSH_VERSION_TIMESTAMP = "push.version.timestamp";
    public final static String PUSH_BACKOFF_DELAY_SECONDS = "push.backoff.delay.seconds";
    public final static String PUSH_ROLLBACK = "push.rollback";
    public final static String PUSH_FORCE_SCHEMA_KEY = "push.force.schema.key";
    public final static String PUSH_FORCE_SCHEMA_VALUE = "push.force.schema.value";
    // others.optional
    public final static String KEY_SELECTION = "key.selection";
    public final static String VALUE_SELECTION = "value.selection";
    public final static String BUILD = "build";
    public final static String PUSH = "push";
    public final static String FETCH_ALL_STORES_XML = "fetch.all.stores.xml";
    public final static String VOLDEMORT_FETCHER_PROTOCOL = "voldemort.fetcher.protocol";
    public final static String VOLDEMORT_FETCHER_PORT = "voldemort.fetcher.port";
    public final static String AVRO_SERIALIZER_VERSIONED = "avro.serializer.versioned";
    public final static String AVRO_KEY_FIELD = "avro.key.field";
    public final static String AVRO_VALUE_FIELD = "avro.value.field";
    public final static String HADOOP_JOB_UGI = "hadoop.job.ugi";
    public final static String REDUCER_PER_BUCKET = "reducer.per.bucket";
    public final static String CHECKSUM_TYPE = "checksum.type";
    public final static String SAVE_KEYS = "save.keys";
    public final static String HEARTBEAT_HOOK_INTERVAL_MS = "heartbeat.hook.interval.ms";
    public final static String HOOKS = "hooks";
    public final static String MIN_NUMBER_OF_RECORDS = "min.number.of.records";
    public final static String REDUCER_OUTPUT_COMPRESS_CODEC = "reducer.output.compress.codec";
    public final static String REDUCER_OUTPUT_COMPRESS = "reducer.output.compress";
    public final static String STORE_VERIFICATION_MAX_THREAD_NUM = "store.verification.max.thread.num";
    // default
    private final static String RECOMMENDED_FETCHER_PROTOCOL = "webhdfs";
    private final int DEFAULT_THREAD_NUM_FOR_STORE_VERIFICATION = 20;

    // CONFIG VALUES (and other immutable state)
    private final Props props;
    private final String storeName;
    private final List<String> clusterURLs;
    private final Map<String, AdminClient> adminClientPerCluster;
    private final List<String> dataDirs;
    private final boolean isAvroJob;
    private final String keyFieldName;
    private final String valueFieldName;
    private final boolean isAvroVersioned;
    private final long minNumberOfRecords;
    private final String hdfsFetcherPort;
    private final String hdfsFetcherProtocol;
    private final String jsonKeyField;
    private final String jsonValueField;
    private final String description;
    private final String owners;
    private final Set<BuildAndPushHook> hooks = new HashSet<BuildAndPushHook>();
    private final int heartBeatHookIntervalTime;
    private final HeartBeatHookRunnable heartBeatHookRunnable;
    private final boolean pushHighAvailability;
    private final List<Closeable> closeables = Lists.newArrayList();
    private final ExecutorService executorService;
    private final boolean enableStoreCreation;
    // Executor to do store/schema verification in parallel
    private final int maxThreadNumForStoreVerification;
    private ExecutorService storeVerificationExecutorService;
    private final HashMap<String, Exception> exceptions = Maps.newHashMap();

    // Mutable state
    private StoreDefinition storeDef;
    private Path sanitizedInputPath = null;
    private Schema inputPathAvroSchema = null;
    private JsonSchema inputPathJsonSchema = null;
    private Future heartBeatHookFuture = null;
    private Map<String, VAdminProto.GetHighAvailabilitySettingsResponse> haSettingsPerCluster;
    private boolean buildPrimaryReplicasOnly;

    private AdminClient createAdminClient(String url, boolean fetchAllStoresXml) {
        ClientConfig config = new ClientConfig().setBootstrapUrls(url)
                .setConnectionTimeout(15,TimeUnit.SECONDS)
                .setFetchAllStoresXmlInBootstrap(fetchAllStoresXml);

        AdminClientConfig adminConfig = new AdminClientConfig().setAdminSocketTimeoutSec(60);
        return new AdminClient(adminConfig, config);
    }

    public VoldemortBuildAndPushJob(String name, azkaban.utils.Props azkabanProps) {
        super(name, Logger.getLogger(name));
        this.log = getLog();

        this.props = new Props(azkabanProps.toProperties());
        log.info("Job props:\n" + this.props.toString(true));

        this.storeName = props.getString(PUSH_STORE_NAME).trim();
        this.clusterURLs = new ArrayList<String>();
        this.dataDirs = new ArrayList<String>();
        this.adminClientPerCluster = Maps.newHashMap();

        String clusterUrlText = props.getString(PUSH_CLUSTER);
        boolean fetchAllStoresXml = props.getBoolean(FETCH_ALL_STORES_XML, true);
        for(String url: Utils.COMMA_SEP.split(clusterUrlText.trim())) {
            if(url.trim().length() > 0) {
                if (clusterURLs.contains(url)) {
                    throw new VoldemortException("the URL: " + url + " is repeated in the "+ PUSH_CLUSTER + " property ");
                }
                try {
                    AdminClient adminClient = createAdminClient(url, fetchAllStoresXml);
                    this.clusterURLs.add(url);
                    this.adminClientPerCluster.put(url, adminClient);
                    this.closeables.add(adminClient);
                } catch (Exception e) {
                    if (ExceptionUtils.recursiveClassEquals(e, BootstrapFailureException.class)) {
                        this.log.error("Unable to reach cluster: " + url + " ... this cluster will be skipped.", e);

                        // We add the exception in the exceptions map so that the job fails after pushing to the healthy clusters
                        exceptions.put(url, new VoldemortException(<commit_msg>Made admin connection/socket timeout configurable in BnP.<commit_after>/*
 * Copyright 2008-2013 LinkedIn, 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 voldemort.store.readonly.mr.azkaban;

import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.apache.avro.Schema;
import org.apache.avro.mapred.AvroInputFormat;
import org.apache.commons.lang.Validate;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.log4j.Logger;

import voldemort.VoldemortException;
import voldemort.client.BootstrapFailureException;
import voldemort.client.ClientConfig;
import voldemort.client.protocol.admin.AdminClient;
import voldemort.client.protocol.admin.AdminClientConfig;
import voldemort.client.protocol.pb.VAdminProto;
import voldemort.cluster.Cluster;
import voldemort.cluster.Node;
import voldemort.serialization.DefaultSerializerFactory;
import voldemort.server.VoldemortConfig;
import voldemort.store.StoreDefinition;
import voldemort.store.UnreachableStoreException;
import voldemort.store.readonly.checksum.CheckSum;
import voldemort.store.readonly.checksum.CheckSum.CheckSumType;
import voldemort.store.readonly.disk.HadoopStoreWriter;
import voldemort.store.readonly.disk.KeyValueWriter;
import voldemort.store.readonly.hooks.BuildAndPushHook;
import voldemort.store.readonly.hooks.BuildAndPushStatus;
import voldemort.store.readonly.mr.AbstractStoreBuilderConfigurable;
import voldemort.store.readonly.mr.AvroStoreBuilderMapper;
import voldemort.store.readonly.mr.HadoopStoreBuilder;
import voldemort.store.readonly.mr.JsonStoreBuilderMapper;
import voldemort.store.readonly.mr.serialization.JsonSequenceFileInputFormat;
import voldemort.store.readonly.mr.utils.AvroUtils;
import voldemort.store.readonly.mr.utils.HadoopUtils;
import voldemort.store.readonly.mr.utils.JsonSchema;
import voldemort.store.readonly.mr.utils.VoldemortUtils;
import voldemort.store.readonly.swapper.DeleteAllFailedFetchStrategy;
import voldemort.store.readonly.swapper.DisableStoreOnFailedNodeFailedFetchStrategy;
import voldemort.store.readonly.swapper.FailedFetchStrategy;
import voldemort.store.readonly.swapper.RecoverableFailedFetchException;
import voldemort.utils.ExceptionUtils;
import voldemort.utils.Props;
import voldemort.utils.ReflectUtils;
import voldemort.utils.Utils;
import azkaban.jobExecutor.AbstractJob;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.protobuf.UninitializedMessageException;

public class VoldemortBuildAndPushJob extends AbstractJob {

    private final Logger log;

    // CONFIG NAME CONSTANTS

    // build.required
    public final static String BUILD_INPUT_PATH = "build.input.path";
    public final static String BUILD_OUTPUT_DIR = "build.output.dir";
    // build.optional
    public final static String BUILD_TEMP_DIR = "build.temp.dir";
    public final static String BUILD_REPLICATION_FACTOR = "build.replication.factor";
    public final static String BUILD_COMPRESS_VALUE = "build.compress.value";
    public final static String BUILD_CHUNK_SIZE = "build.chunk.size";
    public final static String BUILD_OUTPUT_KEEP = "build.output.keep";
    public final static String BUILD_TYPE_AVRO = "build.type.avro";
    public final static String BUILD_REQUIRED_READS = "build.required.reads";
    public final static String BUILD_REQUIRED_WRITES = "build.required.writes";
    public final static String BUILD_FORCE_SCHEMA_KEY = "build.force.schema.key";
    public final static String BUILD_FORCE_SCHEMA_VALUE = "build.force.schema.value";
    public final static String BUILD_PREFERRED_READS = "build.preferred.reads";
    public final static String BUILD_PREFERRED_WRITES = "build.preferred.writes";
    public final static String BUILD_PRIMARY_REPLICAS_ONLY = "build.primary.replicas.only";
    // push.required
    public final static String PUSH_STORE_NAME = "push.store.name";
    public final static String PUSH_CLUSTER = "push.cluster";
    public final static String PUSH_STORE_OWNERS = "push.store.owners";
    public final static String PUSH_STORE_DESCRIPTION = "push.store.description";
    public final static String ENABLE_STORE_CREATION = "enable.store.creation";
    // push.optional
    public final static String PUSH_HTTP_TIMEOUT_SECONDS = "push.http.timeout.seconds";
    public final static String PUSH_VERSION = "push.version";
    public final static String PUSH_VERSION_TIMESTAMP = "push.version.timestamp";
    public final static String PUSH_BACKOFF_DELAY_SECONDS = "push.backoff.delay.seconds";
    public final static String PUSH_ROLLBACK = "push.rollback";
    public final static String PUSH_FORCE_SCHEMA_KEY = "push.force.schema.key";
    public final static String PUSH_FORCE_SCHEMA_VALUE = "push.force.schema.value";
    // others.optional
    public final static String KEY_SELECTION = "key.selection";
    public final static String VALUE_SELECTION = "value.selection";
    public final static String BUILD = "build";
    public final static String PUSH = "push";
    public final static String FETCH_ALL_STORES_XML = "fetch.all.stores.xml";
    public final static String VOLDEMORT_FETCHER_PROTOCOL = "voldemort.fetcher.protocol";
    public final static String VOLDEMORT_FETCHER_PORT = "voldemort.fetcher.port";
    public final static String AVRO_SERIALIZER_VERSIONED = "avro.serializer.versioned";
    public final static String AVRO_KEY_FIELD = "avro.key.field";
    public final static String AVRO_VALUE_FIELD = "avro.value.field";
    public final static String HADOOP_JOB_UGI = "hadoop.job.ugi";
    public final static String REDUCER_PER_BUCKET = "reducer.per.bucket";
    public final static String CHECKSUM_TYPE = "checksum.type";
    public final static String SAVE_KEYS = "save.keys";
    public final static String HEARTBEAT_HOOK_INTERVAL_MS = "heartbeat.hook.interval.ms";
    public final static String HOOKS = "hooks";
    public final static String MIN_NUMBER_OF_RECORDS = "min.number.of.records";
    public final static String REDUCER_OUTPUT_COMPRESS_CODEC = "reducer.output.compress.codec";
    public final static String REDUCER_OUTPUT_COMPRESS = "reducer.output.compress";
    public final static String STORE_VERIFICATION_MAX_THREAD_NUM = "store.verification.max.thread.num";
    public final static String ADMIN_CLIENT_CONNECTION_TIMEOUT_SEC = "admin.client.connection.timeout.sec";
    public final static String ADMIN_CLIENT_SOCKET_TIMEOUT_SEC = "admin.client.socket.timeout.sec";
    // default
    private final static String RECOMMENDED_FETCHER_PROTOCOL = "webhdfs";
    private final int DEFAULT_THREAD_NUM_FOR_STORE_VERIFICATION = 20;

    // CONFIG VALUES (and other immutable state)
    private final Props props;
    private final String storeName;
    private final List<String> clusterURLs;
    private final Map<String, AdminClient> adminClientPerCluster;
    private final List<String> dataDirs;
    private final boolean isAvroJob;
    private final String keyFieldName;
    private final String valueFieldName;
    private final boolean isAvroVersioned;
    private final long minNumberOfRecords;
    private final String hdfsFetcherPort;
    private final String hdfsFetcherProtocol;
    private final String jsonKeyField;
    private final String jsonValueField;
    private final String description;
    private final String owners;
    private final Set<BuildAndPushHook> hooks = new HashSet<BuildAndPushHook>();
    private final int heartBeatHookIntervalTime;
    private final HeartBeatHookRunnable heartBeatHookRunnable;
    private final boolean pushHighAvailability;
    private final List<Closeable> closeables = Lists.newArrayList();
    private final ExecutorService executorService;
    private final boolean enableStoreCreation;
    // Executor to do store/schema verification in parallel
    private final int maxThreadNumForStoreVerification;
    private ExecutorService storeVerificationExecutorService;
    private final HashMap<String, Exception> exceptions = Maps.newHashMap();

    // Mutable state
    private StoreDefinition storeDef;
    private Path sanitizedInputPath = null;
    private Schema inputPathAvroSchema = null;
    private JsonSchema inputPathJsonSchema = null;
    private Future heartBeatHookFuture = null;
    private Map<String, VAdminProto.GetHighAvailabilitySettingsResponse> haSettingsPerCluster;
    private boolean buildPrimaryReplicasOnly;

    private AdminClient createAdminClient(String url, boolean fetchAllStoresXml, int connectionTimeoutSec, int socketTimeoutSec) {
        ClientConfig config = new ClientConfig().setBootstrapUrls(url)
                .setConnectionTimeout(connectionTimeoutSec ,TimeUnit.SECONDS)
                .setFetchAllStoresXmlInBootstrap(fetchAllStoresXml);

        AdminClientConfig adminConfig = new AdminClientConfig().setAdminSocketTimeoutSec(socketTimeoutSec);
        return new AdminClient(adminConfig, config);
    }

    public VoldemortBuildAndPushJob(String name, azkaban.utils.Props azkabanProps) {
        super(name, Logger.getLogger(name));
        this.log = getLog();

        this.props = new Props(azkabanProps.toProperties());
        log.info("Job props:\n" + this.props.toString(true));

        this.storeName = props.getString(PUSH_STORE_NAME).trim();
        final int connectionTimeoutSec = props.getInt(ADMIN_CLIENT_CONNECTION_TIMEOUT_SEC, 15);
        final int socketTimeoutSec = props.getInt(ADMIN_CLIENT_SOCKET_TIMEOUT_SEC, 180);
        this.clusterURLs = new ArrayList<String>();
        this.dataDirs = new ArrayList<String>();
        this.adminClientPerCluster = Maps.newHashMap();

        String clusterUrlText = props.getString(PUSH_CLUSTER);
        boolean fetchAllStoresXml = props.getBoolean(FETCH_ALL_STORES_XML, true);
        for(String url: Utils.COMMA_SEP.split(clusterUrlText.trim())) {
            if(url.trim().length() > 0) {
                if (clusterURLs.contains(url)) {
                    throw new VoldemortException("the URL: " + url + " is repeated in the "+ PUSH_CLUSTER + " property ");
                }
                try {
                    AdminClient adminClient = createAdminClient(url, fetchAllStoresXml, connectionTimeoutSec, socketTimeoutSec);
                    this.clusterURLs.add(url);
                    this.adminClientPerCluster.put(url, adminClient);
                    this.closeables.add(adminClient);
                } catch (Exception e) {
                    if (ExceptionUtils.recursiveClassEquals(e, BootstrapFailureException.class)) {
                        this.log.error("Unable to reach cluster: " + url + " ... this cluster will be skipped.", e);

                        // We add the exception in the exceptions map so that the job fails after pushing to the healthy clusters
                        exceptions.put(url, new VoldemortException(
{'fraction_non_alphanumeric': 0.0800376647834275, 'fraction_numerical': 0.0017548365005992125, 'mean_word_length': 4.368795955882353, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '382313', 'n_tokens_mistral': 6790, 'n_tokens_neox': 6638, 'n_words': 1630}