<commit_before>/*
 * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
 * or more contributor license agreements. Licensed under the Elastic License;
 * you may not use this file except in compliance with the Elastic License.
 */
package org.elasticsearch.alerts.actions;

import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.ElasticsearchIllegalArgumentException;
import org.elasticsearch.ElasticsearchIllegalStateException;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.alerts.*;
import org.elasticsearch.alerts.plugin.AlertsPlugin;
import org.elasticsearch.alerts.triggers.TriggerManager;
import org.elasticsearch.alerts.triggers.TriggerResult;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.hppc.cursors.ObjectCursor;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.joda.time.DateTime;
import org.elasticsearch.common.joda.time.format.DateTimeFormat;
import org.elasticsearch.common.joda.time.format.DateTimeFormatter;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.threadpool.ThreadPool;

import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/**
 */
public class AlertActionManager extends AbstractComponent {

    public static final String ALERT_NAME_FIELD = "alert_name";
    public static final String TRIGGERED_FIELD = "triggered";
    public static final String FIRE_TIME_FIELD = "fire_time";
    public static final String SCHEDULED_FIRE_TIME_FIELD = "scheduled_fire_time";
    public static final String ERROR_MESSAGE = "error_msg";
    public static final String TRIGGER_FIELD = "trigger";
    public static final String REQUEST = "request";
    public static final String RESPONSE = "response";
    public static final String ACTIONS_FIELD = "actions";
    public static final String STATE = "state";

    public static final String ALERT_HISTORY_INDEX_PREFIX = ".alert_history_";
    public static final DateTimeFormatter alertHistoryIndexTimeFormat = DateTimeFormat.forPattern("YYYY-MM-dd");
    public static final String ALERT_HISTORY_TYPE = "alerthistory";

    private static AlertActionEntry END_ENTRY = new AlertActionEntry();

    private final Client client;
    private AlertManager alertManager;
    private final ThreadPool threadPool;
    private final AlertsStore alertsStore;
    private final TriggerManager triggerManager;
    private final TemplateHelper templateHelper;
    private final AlertActionRegistry actionRegistry;

    private final int scrollSize;
    private final TimeValue scrollTimeout;

    private final AtomicLong largestQueueSize = new AtomicLong(0);
    private final AtomicBoolean started = new AtomicBoolean(false);
    private final BlockingQueue<AlertActionEntry> actionsToBeProcessed = new LinkedBlockingQueue<>();

    @Inject
    public AlertActionManager(Settings settings, Client client, AlertActionRegistry actionRegistry,
                              ThreadPool threadPool, AlertsStore alertsStore, TriggerManager triggerManager,
                              TemplateHelper templateHelper) {
        super(settings);
        this.client = client;
        this.actionRegistry = actionRegistry;
        this.threadPool = threadPool;
        this.alertsStore = alertsStore;
        this.triggerManager = triggerManager;
        this.templateHelper = templateHelper;
        // Not using component settings, to let AlertsStore and AlertActionManager share the same settings
        this.scrollSize = settings.getAsInt("alerts.scroll.size", 100);
        this.scrollTimeout = settings.getAsTime("alerts.scroll.timeout", TimeValue.timeValueSeconds(30));
    }

    public void setAlertManager(AlertManager alertManager){
        this.alertManager = alertManager;
    }

    public boolean start(ClusterState state) {
        if (started.get()) {
            return true;
        }
        String[] indices = state.metaData().concreteIndices(IndicesOptions.lenientExpandOpen(), ALERT_HISTORY_INDEX_PREFIX + "*");
        if (indices.length == 0) {
            logger.info("No previous .alerthistory index, skip loading of alert actions");
            templateHelper.checkAndUploadIndexTemplate(state, "alerthistory");
            doStart();
            return true;
        }


        for (String index : indices) {
            IndexMetaData indexMetaData = state.getMetaData().index(index);
            if (indexMetaData != null) {
                if (!state.routingTable().index(index).allPrimaryShardsActive()) {
                    logger.info("Not all primary shards of the [{}] index are started", index);
                    return false;
                }
            }
        }

        try {
            loadQueue();
        } catch (Exception e) {
            logger.error("Unable to load unfinished jobs into the job queue", e);
            actionsToBeProcessed.clear();
        }
        templateHelper.checkAndUploadIndexTemplate(state, "alerthistory");
        doStart();
        return true;
    }

    public void stop() {
        if (started.compareAndSet(true, false)) {
            actionsToBeProcessed.clear();
            actionsToBeProcessed.add(END_ENTRY);
            logger.info("Stopped job queue");
        }
    }

    public boolean started() {
        return started.get();
    }

    /**
     * Calculates the correct alert history index name for a given time using alertHistoryIndexTimeFormat
<<<<<<< HEAD
=======
     * @param time
     * @return
>>>>>>> 7462f14f6d6a8c1529fc4f4203184f30c83057d7
     */
    public static String getAlertHistoryIndexNameForTime(DateTime time) {
        StringBuffer sb = new StringBuffer(ALERT_HISTORY_INDEX_PREFIX);
        alertHistoryIndexTimeFormat.printTo(sb, time);
        return sb.toString();
    }

    private void doStart() {
        logger.info("Starting job queue");
        if (started.compareAndSet(false, true)) {
            threadPool.executor(ThreadPool.Names.GENERIC).execute(new QueueReaderThread());<commit_msg>Remove merge conflict in comment.<commit_after>/*
 * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
 * or more contributor license agreements. Licensed under the Elastic License;
 * you may not use this file except in compliance with the Elastic License.
 */
package org.elasticsearch.alerts.actions;

import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.ElasticsearchIllegalArgumentException;
import org.elasticsearch.ElasticsearchIllegalStateException;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.alerts.*;
import org.elasticsearch.alerts.plugin.AlertsPlugin;
import org.elasticsearch.alerts.triggers.TriggerManager;
import org.elasticsearch.alerts.triggers.TriggerResult;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.joda.time.DateTime;
import org.elasticsearch.common.joda.time.format.DateTimeFormat;
import org.elasticsearch.common.joda.time.format.DateTimeFormatter;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.threadpool.ThreadPool;

import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/**
 */
public class AlertActionManager extends AbstractComponent {

    public static final String ALERT_NAME_FIELD = "alert_name";
    public static final String TRIGGERED_FIELD = "triggered";
    public static final String FIRE_TIME_FIELD = "fire_time";
    public static final String SCHEDULED_FIRE_TIME_FIELD = "scheduled_fire_time";
    public static final String ERROR_MESSAGE = "error_msg";
    public static final String TRIGGER_FIELD = "trigger";
    public static final String REQUEST = "request";
    public static final String RESPONSE = "response";
    public static final String ACTIONS_FIELD = "actions";
    public static final String STATE = "state";

    public static final String ALERT_HISTORY_INDEX_PREFIX = ".alert_history_";
    public static final DateTimeFormatter alertHistoryIndexTimeFormat = DateTimeFormat.forPattern("YYYY-MM-dd");
    public static final String ALERT_HISTORY_TYPE = "alerthistory";

    private static AlertActionEntry END_ENTRY = new AlertActionEntry();

    private final Client client;
    private AlertManager alertManager;
    private final ThreadPool threadPool;
    private final AlertsStore alertsStore;
    private final TriggerManager triggerManager;
    private final TemplateHelper templateHelper;
    private final AlertActionRegistry actionRegistry;

    private final int scrollSize;
    private final TimeValue scrollTimeout;

    private final AtomicLong largestQueueSize = new AtomicLong(0);
    private final AtomicBoolean started = new AtomicBoolean(false);
    private final BlockingQueue<AlertActionEntry> actionsToBeProcessed = new LinkedBlockingQueue<>();

    @Inject
    public AlertActionManager(Settings settings, Client client, AlertActionRegistry actionRegistry,
                              ThreadPool threadPool, AlertsStore alertsStore, TriggerManager triggerManager,
                              TemplateHelper templateHelper) {
        super(settings);
        this.client = client;
        this.actionRegistry = actionRegistry;
        this.threadPool = threadPool;
        this.alertsStore = alertsStore;
        this.triggerManager = triggerManager;
        this.templateHelper = templateHelper;
        // Not using component settings, to let AlertsStore and AlertActionManager share the same settings
        this.scrollSize = settings.getAsInt("alerts.scroll.size", 100);
        this.scrollTimeout = settings.getAsTime("alerts.scroll.timeout", TimeValue.timeValueSeconds(30));
    }

    public void setAlertManager(AlertManager alertManager){
        this.alertManager = alertManager;
    }

    public boolean start(ClusterState state) {
        if (started.get()) {
            return true;
        }
        String[] indices = state.metaData().concreteIndices(IndicesOptions.lenientExpandOpen(), ALERT_HISTORY_INDEX_PREFIX + "*");
        if (indices.length == 0) {
            logger.info("No previous .alerthistory index, skip loading of alert actions");
            templateHelper.checkAndUploadIndexTemplate(state, "alerthistory");
            doStart();
            return true;
        }


        for (String index : indices) {
            IndexMetaData indexMetaData = state.getMetaData().index(index);
            if (indexMetaData != null) {
                if (!state.routingTable().index(index).allPrimaryShardsActive()) {
                    logger.info("Not all primary shards of the [{}] index are started", index);
                    return false;
                }
            }
        }

        try {
            loadQueue();
        } catch (Exception e) {
            logger.error("Unable to load unfinished jobs into the job queue", e);
            actionsToBeProcessed.clear();
        }
        templateHelper.checkAndUploadIndexTemplate(state, "alerthistory");
        doStart();
        return true;
    }

    public void stop() {
        if (started.compareAndSet(true, false)) {
            actionsToBeProcessed.clear();
            actionsToBeProcessed.add(END_ENTRY);
            logger.info("Stopped job queue");
        }
    }

    public boolean started() {
        return started.get();
    }

    /**
     * Calculates the correct alert history index name for a given time using alertHistoryIndexTimeFormat
     */
    public static String getAlertHistoryIndexNameForTime(DateTime time) {
        StringBuffer sb = new StringBuffer(ALERT_HISTORY_INDEX_PREFIX);
        alertHistoryIndexTimeFormat.printTo(sb, time);
        return sb.toString();
    }

    private void doStart() {
        logger.info("Starting job queue");
        if (started.compareAndSet(false, true)) {
            threadPool.executor(ThreadPool.Names.GENERIC).execute(new QueueReaderThread());
{'fraction_non_alphanumeric': 0.0838654135877753, 'fraction_numerical': 0.0030848697897984073, 'mean_word_length': 3.6825663419549883, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 14, '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': '5028661', 'n_tokens_mistral': 3563, 'n_tokens_neox': 3533, 'n_words': 870}