<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.xpack.watcher;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.elasticsearch.ElasticsearchParseException;
import org.elasticsearch.cluster.ClusterChangedEvent;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.ClusterStateListener;
import org.elasticsearch.cluster.block.ClusterBlockLevel;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.routing.AllocationId;
import org.elasticsearch.cluster.routing.IndexRoutingTable;
import org.elasticsearch.cluster.routing.Murmur3HashFunction;
import org.elasticsearch.cluster.routing.RoutingNode;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.set.Sets;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.engine.Engine;
import org.elasticsearch.index.shard.IndexingOperationListener;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.xpack.core.watcher.watch.Watch;
import org.elasticsearch.xpack.watcher.trigger.TriggerService;
import org.elasticsearch.xpack.watcher.watch.WatchParser;
import org.elasticsearch.xpack.watcher.watch.WatchStoreUtils;
import org.joda.time.DateTime;

import java.io.IOException;
import java.time.Clock;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import static org.elasticsearch.cluster.routing.ShardRoutingState.RELOCATING;
import static org.elasticsearch.cluster.routing.ShardRoutingState.STARTED;
import static org.joda.time.DateTimeZone.UTC;

/**
 * This index listener ensures, that watches that are being indexed are put into the trigger service
 * Because the condition for this might change based on the shard allocation, this class is also a
 * cluster state listener
 *
 * Whenever a write operation to the current active watch index is made, this listener checks, if
 * the document should also be added to the local trigger service
 *
 */
final class WatcherIndexingListener extends AbstractComponent implements IndexingOperationListener, ClusterStateListener {

    static final Configuration INACTIVE = new Configuration(null, Collections.emptyMap());

    private final WatchParser parser;
    private final Clock clock;
    private final TriggerService triggerService;
    private volatile Configuration configuration = INACTIVE;

    WatcherIndexingListener(Settings settings, WatchParser parser, Clock clock, TriggerService triggerService) {
        super(settings);
        this.parser = parser;
        this.clock = clock;
        this.triggerService = triggerService;
    }

    // package private for testing
    Configuration getConfiguration() {
        return configuration;
    }

    // package private for testing
    void setConfiguration(Configuration configuration) {
        this.configuration = configuration;
    }

    /**
     * single watch operations that check if the local trigger service should trigger for this
     * concrete watch
     *
     * Watch parsing could be optimized, so that parsing only happens on primary and where the
     * shard is supposed to be put into the trigger service at some point, right no we dont care
     *
     * Note, we have to parse on the primary, because otherwise a failure on the replica when
     * parsing the watch would result in failing
     * the replica
     *
     * @param shardId   The shard id object of the document being processed
     * @param operation The index operation
     * @return          The index operation
     */
    @Override
    public Engine.Index preIndex(ShardId shardId, Engine.Index operation) {
        if (isWatchDocument(shardId.getIndexName(), operation.type())) {
            DateTime now = new DateTime(clock.millis(), UTC);
            try {
                Watch watch = parser.parseWithSecrets(operation.id(), true, operation.source(), now, XContentType.JSON);
                ShardAllocationConfiguration shardAllocationConfiguration = configuration.localShards.get(shardId);
                if (shardAllocationConfiguration == null) {
                    logger.debug("no distributed watch execution info found for watch [{}] on shard [{}], got configuration for {}",
                            watch.id(), shardId, configuration.localShards.keySet());
                    return operation;
                }

                // the watch status is -1, in case a watch has been freshly stored and this save
                // watch operation does not stem from an execution
                // we dont need to update the trigger service, when the watch has been updated as
                // part of an execution, so we can exit early
                boolean isWatchExecutionOperation = watch.status().version() != -1;
                if (isWatchExecutionOperation) {
                    logger.debug("not updating trigger for watch [{}], watch has been updated as part of an execution", watch.id());
                    return operation;
                }

                boolean shouldBeTriggered = shardAllocationConfiguration.shouldBeTriggered(watch.id());
                if (shouldBeTriggered) {
                    if (watch.status().state().isActive()) {
                        logger.debug("adding watch [{}] to trigger service", watch.id());
                        triggerService.add(watch);
                    } else {
                        logger.debug("removing watch [{}] to trigger service", watch.id());
                        triggerService.remove(watch.id());
                    }
                } else {
                    logger.debug("watch [{}] should not be triggered", watch.id());
                }
            } catch (IOException e) {
                throw new ElasticsearchParseException("Could not parse watch with id [{}]", e, operation.id());
            }

        }

        return operation;
    }
<commit_msg>Watcher: Remove never executed code (#31135)<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.xpack.watcher;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.elasticsearch.ElasticsearchParseException;
import org.elasticsearch.cluster.ClusterChangedEvent;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.ClusterStateListener;
import org.elasticsearch.cluster.block.ClusterBlockLevel;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.routing.AllocationId;
import org.elasticsearch.cluster.routing.IndexRoutingTable;
import org.elasticsearch.cluster.routing.Murmur3HashFunction;
import org.elasticsearch.cluster.routing.RoutingNode;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.set.Sets;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.engine.Engine;
import org.elasticsearch.index.shard.IndexingOperationListener;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.xpack.core.watcher.watch.Watch;
import org.elasticsearch.xpack.watcher.trigger.TriggerService;
import org.elasticsearch.xpack.watcher.watch.WatchParser;
import org.elasticsearch.xpack.watcher.watch.WatchStoreUtils;
import org.joda.time.DateTime;

import java.io.IOException;
import java.time.Clock;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import static org.elasticsearch.cluster.routing.ShardRoutingState.RELOCATING;
import static org.elasticsearch.cluster.routing.ShardRoutingState.STARTED;
import static org.joda.time.DateTimeZone.UTC;

/**
 * This index listener ensures, that watches that are being indexed are put into the trigger service
 * Because the condition for this might change based on the shard allocation, this class is also a
 * cluster state listener
 *
 * Whenever a write operation to the current active watch index is made, this listener checks, if
 * the document should also be added to the local trigger service
 *
 */
final class WatcherIndexingListener extends AbstractComponent implements IndexingOperationListener, ClusterStateListener {

    static final Configuration INACTIVE = new Configuration(null, Collections.emptyMap());

    private final WatchParser parser;
    private final Clock clock;
    private final TriggerService triggerService;
    private volatile Configuration configuration = INACTIVE;

    WatcherIndexingListener(Settings settings, WatchParser parser, Clock clock, TriggerService triggerService) {
        super(settings);
        this.parser = parser;
        this.clock = clock;
        this.triggerService = triggerService;
    }

    // package private for testing
    Configuration getConfiguration() {
        return configuration;
    }

    // package private for testing
    void setConfiguration(Configuration configuration) {
        this.configuration = configuration;
    }

    /**
     * single watch operations that check if the local trigger service should trigger for this
     * concrete watch
     *
     * Watch parsing could be optimized, so that parsing only happens on primary and where the
     * shard is supposed to be put into the trigger service at some point, right no we dont care
     *
     * Note, we have to parse on the primary, because otherwise a failure on the replica when
     * parsing the watch would result in failing
     * the replica
     *
     * @param shardId   The shard id object of the document being processed
     * @param operation The index operation
     * @return          The index operation
     */
    @Override
    public Engine.Index preIndex(ShardId shardId, Engine.Index operation) {
        if (isWatchDocument(shardId.getIndexName(), operation.type())) {
            DateTime now = new DateTime(clock.millis(), UTC);
            try {
                Watch watch = parser.parseWithSecrets(operation.id(), true, operation.source(), now, XContentType.JSON);
                ShardAllocationConfiguration shardAllocationConfiguration = configuration.localShards.get(shardId);
                if (shardAllocationConfiguration == null) {
                    logger.debug("no distributed watch execution info found for watch [{}] on shard [{}], got configuration for {}",
                            watch.id(), shardId, configuration.localShards.keySet());
                    return operation;
                }

                boolean shouldBeTriggered = shardAllocationConfiguration.shouldBeTriggered(watch.id());
                if (shouldBeTriggered) {
                    if (watch.status().state().isActive()) {
                        logger.debug("adding watch [{}] to trigger service", watch.id());
                        triggerService.add(watch);
                    } else {
                        logger.debug("removing watch [{}] to trigger service", watch.id());
                        triggerService.remove(watch.id());
                    }
                } else {
                    logger.debug("watch [{}] should not be triggered", watch.id());
                }
            } catch (IOException e) {
                throw new ElasticsearchParseException("Could not parse watch with id [{}]", e, operation.id());
            }

        }

        return operation;
    }

{'fraction_non_alphanumeric': 0.08103950777202072, 'fraction_numerical': 0.0008905440414507772, 'mean_word_length': 3.1108153078202996, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '4632598', 'n_tokens_mistral': 2988, 'n_tokens_neox': 2987, 'n_words': 934}