<commit_before>package org.codelibs.elasticsearch.configsync.service;

import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicBoolean;

import org.codelibs.elasticsearch.configsync.action.ConfigFileFlushResponse;
import org.codelibs.elasticsearch.configsync.action.ConfigResetSyncResponse;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterService;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.common.Base64;
import org.elasticsearch.common.component.AbstractLifecycleComponent;
import org.elasticsearch.common.component.LifecycleListener;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.io.Streams;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.env.Environment;
import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.threadpool.ThreadPool.Names;
import org.elasticsearch.transport.TransportChannel;
import org.elasticsearch.transport.TransportException;
import org.elasticsearch.transport.TransportRequest;
import org.elasticsearch.transport.TransportRequestHandler;
import org.elasticsearch.transport.TransportResponseHandler;
import org.elasticsearch.transport.TransportService;

import com.google.common.base.Charsets;
import com.google.common.collect.UnmodifiableIterator;

public class ConfigSyncService extends AbstractLifecycleComponent<ConfigSyncService> {

    public static final String ACTION_CONFIG_FLUSH = "internal:indices/config/flush";

    public static final String ACTION_CONFIG_RESET = "internal:indices/config/reset_sync";

    private static final String FILE_MAPPING_JSON = "configsync/file_mapping.json";

    public static final String TIMESTAMP = "@timestamp";

    public static final String CONTENT = "content";

    public static final String PATH = "path";

    private final Client client;

    private final String index;

    private final String type;

    private final String configPath;

    private final ThreadPool threadPool;

    private final String scrollForUpdate;

    private final int sizeForUpdate;

    private Date lastChecked = new Date(0);

    private ConfigFileUpdater configFileUpdater;

    private final ClusterService clusterService;

    private final TransportService transportService;

    private volatile ScheduledFuture<?> scheduledFuture;

    @Inject
    public ConfigSyncService(final Settings settings, final Client client, final ClusterService clusterService,
            final TransportService transportService, final Environment env, final ThreadPool threadPool) {
        super(settings);
        this.client = client;
        this.clusterService = clusterService;
        this.transportService = transportService;
        this.threadPool = threadPool;

        logger.info("Creating ConfigSyncService");

        index = settings.get("configsync.index", ".configsync");
        type = settings.get("configsync.type", "file");
        configPath = settings.get("configsync.config_path", env.configFile().toFile().getAbsolutePath());
        scrollForUpdate = settings.get("configsync.scroll_time", "1m");
        sizeForUpdate = settings.getAsInt("configsync.scroll_size", 1);

        transportService.registerRequestHandler(ACTION_CONFIG_FLUSH, FileFlushRequest.class, ThreadPool.Names.GENERIC,
                new ConfigFileFlushRequestHandler());
        transportService.registerRequestHandler(ACTION_CONFIG_RESET, ResetSyncRequest.class, ThreadPool.Names.GENERIC,
                new ConfigSyncResetRequestHandler());
    }

    private void startUpdater() {
        configFileUpdater = new ConfigFileUpdater();

        if (scheduledFuture != null) {
            scheduledFuture.cancel(true);
        }

        final TimeValue flushInterval =
                clusterService.state().getMetaData().settings().getAsTime("configsync.flush_interval", TimeValue.timeValueMinutes(1));
        scheduledFuture = threadPool.schedule(flushInterval, Names.SAME, configFileUpdater);
        if (logger.isDebugEnabled()) {
            logger.debug("Scheduled ConfigFileUpdater with " + flushInterval);
        }
    }

    @Override
    protected void doStart() throws ElasticsearchException {
        logger.info("Starting ConfigSyncService");

        clusterService.addLifecycleListener(new LifecycleListener() {
            @Override
            public void afterStart() {
                client.admin().cluster().prepareHealth().setWaitForYellowStatus().execute(new ActionListener<ClusterHealthResponse>() {

                    @Override
                    public void onResponse(final ClusterHealthResponse response) {
                        if (response.isTimedOut()) {
                            logger.error("Failed to start ConfigSyncService. Elasticsearch was timeouted.");
                        } else {
                            checkIfIndexExists();
                        }
                    }

                    @Override
                    public void onFailure(final Throwable e) {
                        logger.error("Failed to start ConfigSyncService.", e);<commit_msg>skip to schdule ConfigFileUpdater<commit_after>package org.codelibs.elasticsearch.configsync.service;

import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicBoolean;

import org.codelibs.elasticsearch.configsync.action.ConfigFileFlushResponse;
import org.codelibs.elasticsearch.configsync.action.ConfigResetSyncResponse;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterService;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.common.Base64;
import org.elasticsearch.common.component.AbstractLifecycleComponent;
import org.elasticsearch.common.component.LifecycleListener;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.io.Streams;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.env.Environment;
import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.threadpool.ThreadPool.Names;
import org.elasticsearch.transport.TransportChannel;
import org.elasticsearch.transport.TransportException;
import org.elasticsearch.transport.TransportRequest;
import org.elasticsearch.transport.TransportRequestHandler;
import org.elasticsearch.transport.TransportResponseHandler;
import org.elasticsearch.transport.TransportService;

import com.google.common.base.Charsets;
import com.google.common.collect.UnmodifiableIterator;

public class ConfigSyncService extends AbstractLifecycleComponent<ConfigSyncService> {

    public static final String ACTION_CONFIG_FLUSH = "internal:indices/config/flush";

    public static final String ACTION_CONFIG_RESET = "internal:indices/config/reset_sync";

    private static final String FILE_MAPPING_JSON = "configsync/file_mapping.json";

    public static final String TIMESTAMP = "@timestamp";

    public static final String CONTENT = "content";

    public static final String PATH = "path";

    private final Client client;

    private final String index;

    private final String type;

    private final String configPath;

    private final ThreadPool threadPool;

    private final String scrollForUpdate;

    private final int sizeForUpdate;

    private Date lastChecked = new Date(0);

    private ConfigFileUpdater configFileUpdater;

    private final ClusterService clusterService;

    private final TransportService transportService;

    private volatile ScheduledFuture<?> scheduledFuture;

    @Inject
    public ConfigSyncService(final Settings settings, final Client client, final ClusterService clusterService,
            final TransportService transportService, final Environment env, final ThreadPool threadPool) {
        super(settings);
        this.client = client;
        this.clusterService = clusterService;
        this.transportService = transportService;
        this.threadPool = threadPool;

        logger.info("Creating ConfigSyncService");

        index = settings.get("configsync.index", ".configsync");
        type = settings.get("configsync.type", "file");
        configPath = settings.get("configsync.config_path", env.configFile().toFile().getAbsolutePath());
        scrollForUpdate = settings.get("configsync.scroll_time", "1m");
        sizeForUpdate = settings.getAsInt("configsync.scroll_size", 1);

        transportService.registerRequestHandler(ACTION_CONFIG_FLUSH, FileFlushRequest.class, ThreadPool.Names.GENERIC,
                new ConfigFileFlushRequestHandler());
        transportService.registerRequestHandler(ACTION_CONFIG_RESET, ResetSyncRequest.class, ThreadPool.Names.GENERIC,
                new ConfigSyncResetRequestHandler());
    }

    private void startUpdater() {
        configFileUpdater = new ConfigFileUpdater();

        if (scheduledFuture != null) {
            scheduledFuture.cancel(true);
        }

        final TimeValue flushInterval =
                clusterService.state().getMetaData().settings().getAsTime("configsync.flush_interval", TimeValue.timeValueMinutes(1));
        if (flushInterval.millis() < 0) {
            if(logger.isDebugEnabled()){
                logger.debug("ConfigFileUpdater is not scheduled.");
            }
        } else {
            scheduledFuture = threadPool.schedule(flushInterval, Names.SAME, configFileUpdater);
            if (logger.isDebugEnabled()) {
                logger.debug("Scheduled ConfigFileUpdater with " + flushInterval);
            }
        }
    }

    @Override
    protected void doStart() throws ElasticsearchException {
        logger.info("Starting ConfigSyncService");

        clusterService.addLifecycleListener(new LifecycleListener() {
            @Override
            public void afterStart() {
                client.admin().cluster().prepareHealth().setWaitForYellowStatus().execute(new ActionListener<ClusterHealthResponse>() {

                    @Override
                    public void onResponse(final ClusterHealthResponse response) {
                        if (response.isTimedOut()) {
                            logger.error("Failed to start ConfigSyncService. Elasticsearch was timeouted.");
                        } else {
                            checkIfIndexExists();
                        }
                    }

                    @Override
                    public void onFailure(final Throwable e) {
                        logger.error("Failed to start ConfigSyncService.", e);
{'fraction_non_alphanumeric': 0.08652816448918398, 'fraction_numerical': 0.0009281073748839865, 'mean_word_length': 4.01180679785331, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, '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': '1223140', 'n_tokens_mistral': 3463, 'n_tokens_neox': 3475, 'n_words': 695}