<commit_before>// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package ai.vespa.reindexing;

import ai.vespa.reindexing.Reindexing.Status;
import com.google.common.util.concurrent.UncheckedTimeoutException;
import com.yahoo.document.DocumentType;
import com.yahoo.document.DocumentTypeManager;
import com.yahoo.path.Path;
import com.yahoo.slime.Cursor;
import com.yahoo.slime.Inspector;
import com.yahoo.slime.Slime;
import com.yahoo.slime.SlimeUtils;
import com.yahoo.vespa.curator.Curator;
import com.yahoo.vespa.curator.Lock;
import com.yahoo.yolean.Exceptions;

import java.time.Duration;
import java.time.Instant;
import java.util.Map;
import java.util.function.Function;

import static java.util.Objects.requireNonNull;
import static java.util.stream.Collectors.toUnmodifiableMap;

/**
 * Reads and writes status of initiated reindexing jobs.
 *
 * @author jonmv
 */
public class ReindexingCurator {

    private final Curator curator;
    private final ReindexingSerializer serializer;
    private final Duration lockTimeout;

    public ReindexingCurator(Curator curator, DocumentTypeManager manager) {
        this(curator, manager, Duration.ofSeconds(1));
    }

    ReindexingCurator(Curator curator, DocumentTypeManager manager, Duration lockTimeout) {
        this.curator = curator;
        this.serializer = new ReindexingSerializer(manager);
        this.lockTimeout = lockTimeout;
    }

    /** If no reindexing data exists (has been wiped), assume current ready documents are already done. */
    public void initializeIfEmpty(String cluster, Map<DocumentType, Instant> ready, Instant now) {
        if ( ! curator.exists(statusPath(cluster))) {
            try (Lock lock = lockReindexing(cluster)) {
                if (curator.exists(statusPath(cluster)))
                    return; // Some other node already did this.

                Reindexing reindexing = Reindexing.empty();
                for (DocumentType type : ready.keySet())
                    if (ready.get(type).isBefore(now))
                        reindexing = reindexing.with(type, Status.ready(now).running().successful(now));

                writeReindexing(reindexing, cluster);
            }
            catch (ReindexingLockException ignored) {
                // Some other node took ownership and is doing this.
            }
        }
    }

    public Reindexing readReindexing(String cluster) {
        return curator.getData(statusPath(cluster)).map(serializer::deserialize)
                      .orElse(Reindexing.empty());
    }

    public void writeReindexing(Reindexing reindexing, String cluster) {
        curator.set(statusPath(cluster), serializer.serialize(reindexing));
    }

    /** This lock must be held to manipulate reindexing state, or by whoever has a running visitor. */
    public Lock lockReindexing(String cluster) throws ReindexingLockException {
        try {
            return curator.lock(lockPath(cluster), lockTimeout);
        }
        catch (UncheckedTimeoutException e) { // TODO jonmv: Avoid use of guava classes.
            throw new ReindexingLockException(e);
        }
    }

    private Path rootPath(String clusterName) { return Path.fromString("/reindexing/v1/" + clusterName); }
    private Path statusPath(String clusterName) { return rootPath(clusterName).append("status"); }
    private Path lockPath(String clusterName) { return rootPath(clusterName).append("lock"); }


    private static class ReindexingSerializer {

        private static final String STATUS = "status";
        private static final String TYPE =  "type";
        private static final String STARTED_MILLIS = "startedMillis";
        private static final String ENDED_MILLIS = "endedMillis";
        private static final String PROGRESS = "progress";
        private static final String STATE = "state";
        private static final String MESSAGE = "message";
<commit_msg>Add logging to ReindexingCurator<commit_after>// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package ai.vespa.reindexing;

import ai.vespa.reindexing.Reindexing.Status;
import com.google.common.util.concurrent.UncheckedTimeoutException;
import com.yahoo.document.DocumentType;
import com.yahoo.document.DocumentTypeManager;
import com.yahoo.path.Path;
import com.yahoo.slime.Cursor;
import com.yahoo.slime.Inspector;
import com.yahoo.slime.Slime;
import com.yahoo.slime.SlimeUtils;
import com.yahoo.vespa.curator.Curator;
import com.yahoo.vespa.curator.Lock;
import com.yahoo.yolean.Exceptions;

import java.time.Duration;
import java.time.Instant;
import java.util.Map;
import java.util.function.Function;
import java.util.logging.Level;
import java.util.logging.Logger;

import static java.util.Objects.requireNonNull;
import static java.util.stream.Collectors.toUnmodifiableMap;

/**
 * Reads and writes status of initiated reindexing jobs.
 *
 * @author jonmv
 */
public class ReindexingCurator {

    private static final Logger log = Logger.getLogger(ReindexingCurator.class.getName());

    private final Curator curator;
    private final ReindexingSerializer serializer;
    private final Duration lockTimeout;

    public ReindexingCurator(Curator curator, DocumentTypeManager manager) {
        this(curator, manager, Duration.ofSeconds(1));
    }

    ReindexingCurator(Curator curator, DocumentTypeManager manager, Duration lockTimeout) {
        this.curator = curator;
        this.serializer = new ReindexingSerializer(manager);
        this.lockTimeout = lockTimeout;
    }

    /** If no reindexing data exists (has been wiped), assume current ready documents are already done. */
    public void initializeIfEmpty(String cluster, Map<DocumentType, Instant> ready, Instant now) {
        if ( ! curator.exists(statusPath(cluster))) {
            try (Lock lock = lockReindexing(cluster)) {
                if (curator.exists(statusPath(cluster)))
                    return; // Some other node already did this.

                Reindexing reindexing = Reindexing.empty();
                for (DocumentType type : ready.keySet())
                    if (ready.get(type).isBefore(now))
                        reindexing = reindexing.with(type, Status.ready(now).running().successful(now));

                log.log(Level.INFO, "Creating initial reindexing status at '" + statusPath(cluster) + "'");
                writeReindexing(reindexing, cluster);
            }
            catch (ReindexingLockException ignored) {
                // Some other node took ownership and is doing this.
            }
        }
    }

    public Reindexing readReindexing(String cluster) {
        Reindexing reindexing = curator.getData(statusPath(cluster)).map(serializer::deserialize)
                                       .orElse(Reindexing.empty());
        log.log(Level.FINE, () -> "Read reindexing status '" + reindexing + "' from '" + statusPath(cluster) + "'");
        return reindexing;
    }

    public void writeReindexing(Reindexing reindexing, String cluster) {
        log.log(Level.FINE, () -> "Writing reindexing status '" + reindexing + "' to '" + statusPath(cluster) + "'");
        curator.set(statusPath(cluster), serializer.serialize(reindexing));
    }

    /** This lock must be held to manipulate reindexing state, or by whoever has a running visitor. */
    public Lock lockReindexing(String cluster) throws ReindexingLockException {
        try {
            return curator.lock(lockPath(cluster), lockTimeout);
        }
        catch (UncheckedTimeoutException e) { // TODO jonmv: Avoid use of guava classes.
            throw new ReindexingLockException(e);
        }
    }

    private Path rootPath(String clusterName) { return Path.fromString("/reindexing/v1/" + clusterName); }
    private Path statusPath(String clusterName) { return rootPath(clusterName).append("status"); }
    private Path lockPath(String clusterName) { return rootPath(clusterName).append("lock"); }


    private static class ReindexingSerializer {

        private static final String STATUS = "status";
        private static final String TYPE =  "type";
        private static final String STARTED_MILLIS = "startedMillis";
        private static final String ENDED_MILLIS = "endedMillis";
        private static final String PROGRESS = "progress";
        private static final String STATE = "state";
        private static final String MESSAGE = "message";

{'fraction_non_alphanumeric': 0.09393369386315542, 'fraction_numerical': 0.0009405125793557489, 'mean_word_length': 3.1477328132618236, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6964834', 'n_tokens_mistral': 2343, 'n_tokens_neox': 2262, 'n_words': 615}