<commit_before>/*
 * Copyright Strimzi authors.
 * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
 */
package io.strimzi.operator.cluster.operator.assembly;

import io.fabric8.kubernetes.api.model.DeletionPropagation;
import io.fabric8.kubernetes.api.model.LabelSelector;
import io.fabric8.kubernetes.api.model.OwnerReference;
import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.PodBuilder;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientException;
import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
import io.fabric8.kubernetes.client.informers.cache.Lister;
import io.fabric8.kubernetes.client.internal.readiness.Readiness;
import io.strimzi.api.kafka.KafkaList;
import io.strimzi.api.kafka.StrimziPodSetList;
import io.strimzi.api.kafka.model.Kafka;
import io.strimzi.api.kafka.model.StrimziPodSet;
import io.strimzi.api.kafka.model.StrimziPodSetBuilder;
import io.strimzi.api.kafka.model.status.StrimziPodSetStatus;
import io.strimzi.operator.cluster.model.ModelUtils;
import io.strimzi.operator.cluster.model.PodSetUtils;
import io.strimzi.operator.cluster.model.StatusDiff;
import io.strimzi.operator.cluster.operator.resource.PodRevision;
import io.strimzi.operator.common.Reconciliation;
import io.strimzi.operator.common.ReconciliationLogger;
import io.strimzi.operator.common.Util;
import io.strimzi.operator.common.model.Labels;
import io.strimzi.operator.common.operator.resource.CrdOperator;
import io.strimzi.operator.common.operator.resource.PodOperator;
import io.strimzi.operator.common.operator.resource.StatusUtils;

import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.stream.Collectors;

/**
 * StrimziPodSet controller is responsible for managing the StrimziPodSets and the pods which belong to them
 */
public class StrimziPodSetController implements Runnable {
    private static final ReconciliationLogger LOGGER = ReconciliationLogger.create(StrimziPodSetController.class);

    private Thread controllerThread;

    private volatile boolean stop = false;

    private final PodOperator podOperator;
    private final CrdOperator<KubernetesClient, StrimziPodSet, StrimziPodSetList> strimziPodSetOperator;
    private final Optional<LabelSelector> crSelector;
    private final String watchedNamespace;

    private final BlockingQueue<SimplifiedReconciliation> workQueue;
    private final SharedIndexInformer<Pod> podInformer;
    private final SharedIndexInformer<StrimziPodSet> strimziPodSetInformer;
    private final SharedIndexInformer<Kafka> kafkaInformer;
    private final Lister<Pod> podLister;
    private final Lister<StrimziPodSet> strimziPodSetLister;
    private final Lister<Kafka> kafkaLister;

    /**
     * Creates the StrimziPodSet controller. The controller should normally exist once per operator for cluster-wide mode
     * or once per namespace for namespaced mode.
     *
     * @param watchedNamespace      Namespace which should be watched. Use * for all namespaces.
     * @param crSelectorLabels      Selector labels for custom resource managed by this operator instance. This is used
     *                              to check that the pods belong to a Kafka cluster matching these labels.
     * @param kafkaOperator         Kafka Operator for getting the Kafka custom resources
     * @param strimziPodSetOperator StrimziPodSet Operator used to manage the StrimziPodSet resources - get them, update
     *                              their status etc.
     * @param podOperator           Pod operator for managing pods
     */
    public StrimziPodSetController(String watchedNamespace, Labels crSelectorLabels, CrdOperator<KubernetesClient, Kafka, KafkaList> kafkaOperator, CrdOperator<KubernetesClient, StrimziPodSet, StrimziPodSetList> strimziPodSetOperator, PodOperator podOperator, int podSetControllerWorkQueueSize) {
        this.podOperator = podOperator;
        this.strimziPodSetOperator = strimziPodSetOperator;
        this.crSelector = (crSelectorLabels == null || crSelectorLabels.toMap().isEmpty()) ? Optional.empty() : Optional.of(new LabelSelector(null, crSelectorLabels.toMap()));
        this.watchedNamespace = watchedNamespace;
        this.workQueue = new ArrayBlockingQueue<>(podSetControllerWorkQueueSize);

        // Kafka informer and lister is used to get Kafka CRs quickly. This is needed for verification of the CR selector labels
        this.kafkaInformer = kafkaOperator.informer(watchedNamespace, (crSelectorLabels == null) ? Map.of() : crSelectorLabels.toMap());
        this.kafkaLister = new Lister<>(kafkaInformer.getIndexer());

        // StrimziPodSet informer and lister is used to get events about StrimziPodSet and get StrimziPodSet quickly
        this.strimziPodSetInformer = strimziPodSetOperator.informer(watchedNamespace);
        this.strimziPodSetLister = new Lister<>(strimziPodSetInformer.getIndexer());
        this.strimziPodSetInformer.addEventHandlerWithResyncPeriod(new ResourceEventHandler<>() {
            @Override
            public void onAdd(StrimziPodSet podSet) {
                enqueueStrimziPodSet(podSet, "ADDED");
            }

            @Override
            public void onUpdate(StrimziPodSet oldPodSet, StrimziPodSet newPodSet) {
                enqueueStrimziPodSet(newPodSet, "MODIFIED");
            }

            @Override
            public void onDelete(StrimziPodSet podSet, boolean deletedFinalStateUnknown) {
                LOGGER.debugOp("StrimziPodSet {} in namespace {} was {}", podSet.getMetadata().getName(), podSet.getMetadata().getNamespace(), "DELETED");
                // Nothing to do => garbage collection should take care of things<commit_msg>feat: Add missing javadoc param (#6615)<commit_after>/*
 * Copyright Strimzi authors.
 * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
 */
package io.strimzi.operator.cluster.operator.assembly;

import io.fabric8.kubernetes.api.model.DeletionPropagation;
import io.fabric8.kubernetes.api.model.LabelSelector;
import io.fabric8.kubernetes.api.model.OwnerReference;
import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.PodBuilder;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientException;
import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
import io.fabric8.kubernetes.client.informers.cache.Lister;
import io.fabric8.kubernetes.client.internal.readiness.Readiness;
import io.strimzi.api.kafka.KafkaList;
import io.strimzi.api.kafka.StrimziPodSetList;
import io.strimzi.api.kafka.model.Kafka;
import io.strimzi.api.kafka.model.StrimziPodSet;
import io.strimzi.api.kafka.model.StrimziPodSetBuilder;
import io.strimzi.api.kafka.model.status.StrimziPodSetStatus;
import io.strimzi.operator.cluster.model.ModelUtils;
import io.strimzi.operator.cluster.model.PodSetUtils;
import io.strimzi.operator.cluster.model.StatusDiff;
import io.strimzi.operator.cluster.operator.resource.PodRevision;
import io.strimzi.operator.common.Reconciliation;
import io.strimzi.operator.common.ReconciliationLogger;
import io.strimzi.operator.common.Util;
import io.strimzi.operator.common.model.Labels;
import io.strimzi.operator.common.operator.resource.CrdOperator;
import io.strimzi.operator.common.operator.resource.PodOperator;
import io.strimzi.operator.common.operator.resource.StatusUtils;

import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.stream.Collectors;

/**
 * StrimziPodSet controller is responsible for managing the StrimziPodSets and the pods which belong to them
 */
public class StrimziPodSetController implements Runnable {
    private static final ReconciliationLogger LOGGER = ReconciliationLogger.create(StrimziPodSetController.class);

    private Thread controllerThread;

    private volatile boolean stop = false;

    private final PodOperator podOperator;
    private final CrdOperator<KubernetesClient, StrimziPodSet, StrimziPodSetList> strimziPodSetOperator;
    private final Optional<LabelSelector> crSelector;
    private final String watchedNamespace;

    private final BlockingQueue<SimplifiedReconciliation> workQueue;
    private final SharedIndexInformer<Pod> podInformer;
    private final SharedIndexInformer<StrimziPodSet> strimziPodSetInformer;
    private final SharedIndexInformer<Kafka> kafkaInformer;
    private final Lister<Pod> podLister;
    private final Lister<StrimziPodSet> strimziPodSetLister;
    private final Lister<Kafka> kafkaLister;

    /**
     * Creates the StrimziPodSet controller. The controller should normally exist once per operator for cluster-wide mode
     * or once per namespace for namespaced mode.
     *
     * @param watchedNamespace              Namespace which should be watched. Use * for all namespaces.
     * @param crSelectorLabels              Selector labels for custom resource managed by this operator instance. This is used
     *                                      to check that the pods belong to a Kafka cluster matching these labels.
     * @param kafkaOperator                 Kafka Operator for getting the Kafka custom resources
     * @param strimziPodSetOperator         StrimziPodSet Operator used to manage the StrimziPodSet resources - get them, update
     *                                      their status etc.
     * @param podOperator                   Pod operator for managing pods
     * @param podSetControllerWorkQueueSize Indicates the size of the StrimziPodSetController work queue
     */
    public StrimziPodSetController(String watchedNamespace, Labels crSelectorLabels, CrdOperator<KubernetesClient, Kafka, KafkaList> kafkaOperator, CrdOperator<KubernetesClient, StrimziPodSet, StrimziPodSetList> strimziPodSetOperator, PodOperator podOperator, int podSetControllerWorkQueueSize) {
        this.podOperator = podOperator;
        this.strimziPodSetOperator = strimziPodSetOperator;
        this.crSelector = (crSelectorLabels == null || crSelectorLabels.toMap().isEmpty()) ? Optional.empty() : Optional.of(new LabelSelector(null, crSelectorLabels.toMap()));
        this.watchedNamespace = watchedNamespace;
        this.workQueue = new ArrayBlockingQueue<>(podSetControllerWorkQueueSize);

        // Kafka informer and lister is used to get Kafka CRs quickly. This is needed for verification of the CR selector labels
        this.kafkaInformer = kafkaOperator.informer(watchedNamespace, (crSelectorLabels == null) ? Map.of() : crSelectorLabels.toMap());
        this.kafkaLister = new Lister<>(kafkaInformer.getIndexer());

        // StrimziPodSet informer and lister is used to get events about StrimziPodSet and get StrimziPodSet quickly
        this.strimziPodSetInformer = strimziPodSetOperator.informer(watchedNamespace);
        this.strimziPodSetLister = new Lister<>(strimziPodSetInformer.getIndexer());
        this.strimziPodSetInformer.addEventHandlerWithResyncPeriod(new ResourceEventHandler<>() {
            @Override
            public void onAdd(StrimziPodSet podSet) {
                enqueueStrimziPodSet(podSet, "ADDED");
            }

            @Override
            public void onUpdate(StrimziPodSet oldPodSet, StrimziPodSet newPodSet) {
                enqueueStrimziPodSet(newPodSet, "MODIFIED");
            }

            @Override
            public void onDelete(StrimziPodSet podSet, boolean deletedFinalStateUnknown) {
                LOGGER.debugOp("StrimziPodSet {} in namespace {} was {}", podSet.getMetadata().getName(), podSet.getMetadata().getNamespace(), "DELETED");
                // Nothing to do => garbage collection should take care of things
{'fraction_non_alphanumeric': 0.07973909498573176, 'fraction_numerical': 0.0027721157766000815, 'mean_word_length': 4.166807076663859, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6102359', 'n_tokens_mistral': 3487, 'n_tokens_neox': 3391, 'n_words': 794}