<commit_before>// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.vespa.hosted.node.admin.nodeadmin;

import com.yahoo.concurrent.ThreadFactoryFactory;
import com.yahoo.concurrent.classlock.ClassLock;
import com.yahoo.concurrent.classlock.ClassLocking;
import com.yahoo.concurrent.classlock.LockInterruptException;
import com.yahoo.log.LogLevel;
import com.yahoo.vespa.hosted.node.admin.configserver.noderepository.NodeSpec;
import com.yahoo.vespa.hosted.node.admin.configserver.noderepository.NodeRepository;
import com.yahoo.vespa.hosted.node.admin.configserver.orchestrator.Orchestrator;
import com.yahoo.vespa.hosted.node.admin.maintenance.StorageMaintainer;
import com.yahoo.vespa.hosted.node.admin.configserver.noderepository.NodeAttributes;
import com.yahoo.vespa.hosted.node.admin.configserver.orchestrator.OrchestratorException;
import com.yahoo.vespa.hosted.node.admin.provider.NodeAdminStateUpdater;
import com.yahoo.vespa.hosted.node.admin.configserver.HttpException;
import com.yahoo.vespa.hosted.provision.Node;

import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Collectors;

import static com.yahoo.vespa.hosted.node.admin.provider.NodeAdminStateUpdater.State.RESUMED;
import static com.yahoo.vespa.hosted.node.admin.provider.NodeAdminStateUpdater.State.SUSPENDED_NODE_ADMIN;
import static com.yahoo.vespa.hosted.node.admin.provider.NodeAdminStateUpdater.State.TRANSITIONING;

/**
 * Pulls information from node repository and forwards containers to run to node admin.
 *
 * @author dybis, stiankri
 */
public class NodeAdminStateUpdaterImpl implements NodeAdminStateUpdater {
    static final Duration FREEZE_CONVERGENCE_TIMEOUT = Duration.ofMinutes(5);

    private final AtomicBoolean terminated = new AtomicBoolean(false);
    private State currentState = SUSPENDED_NODE_ADMIN;
    private State wantedState = RESUMED;
    private boolean workToDoNow = true;

    private final Object monitor = new Object();

    private final Logger log = Logger.getLogger(NodeAdminStateUpdater.class.getName());
    private final ScheduledExecutorService specVerifierScheduler =
            Executors.newScheduledThreadPool(1, ThreadFactoryFactory.getDaemonThreadFactory("specverifier"));
    private final Thread loopThread;

    private final NodeRepository nodeRepository;
    private final Orchestrator orchestrator;
    private final NodeAdmin nodeAdmin;
    private final Clock clock;
    private final String dockerHostHostName;
    private final Duration nodeAdminConvergeStateInterval;

    private final Optional<ClassLocking> classLocking;
    private Optional<ClassLock> classLock = Optional.empty();
    private Instant lastTick;

    public NodeAdminStateUpdaterImpl(
            NodeRepository nodeRepository,
            Orchestrator orchestrator,
            StorageMaintainer storageMaintainer,
            NodeAdmin nodeAdmin,
            String dockerHostHostName,
            Clock clock,
            Duration nodeAdminConvergeStateInterval,
            Optional<ClassLocking> classLocking) {
        log.info(objectToString() + ": Creating object");
        this.nodeRepository = nodeRepository;
        this.orchestrator = orchestrator;
        this.nodeAdmin = nodeAdmin;
        this.dockerHostHostName = dockerHostHostName;
        this.clock = clock;
        this.nodeAdminConvergeStateInterval = nodeAdminConvergeStateInterval;
        this.classLocking = classLocking;
        this.lastTick = clock.instant();

        this.loopThread = new Thread(() -> {
            if (classLocking.isPresent()) {
                log.info(objectToString() + ": Acquiring lock");
                try {
                    classLock = Optional.of(classLocking.get().lockWhile(NodeAdminStateUpdater.class, () -> !terminated.get()));
                } catch (LockInterruptException e) {
                    classLock = Optional.empty();
                    return;
                }
            }

            log.info(objectToString() + ": Starting threads and schedulers");
            nodeAdmin.start();
            specVerifierScheduler.scheduleWithFixedDelay(() ->
                    updateHardwareDivergence(storageMaintainer), 5, 60, TimeUnit.MINUTES);

            while (! terminated.get()) {
                tick();
            }
        });
        this.loopThread.setName("tick-NodeAdminStateUpdater");
    }

    private String objectToString() {
        return this.getClass().getSimpleName() + "@" + Integer.toString(System.identityHashCode(this));
    }

    @Override
    public Map<String, Object> getDebugPage() {
        Map<String, Object> debug = new LinkedHashMap<>();
        synchronized (monitor) {
            debug.put("dockerHostHostName", dockerHostHostName);
            debug.put("wantedState", wantedState);
            debug.put("currentState", currentState);
            debug.put("NodeAdmin", nodeAdmin.debugInfo());
        }
        return debug;
    }

    private void updateHardwareDivergence(StorageMaintainer maintainer) {
        if (currentState != RESUMED) return;

        try {
            NodeSpec node = nodeRepository.getOptionalNode(dockerHostHostName)
                    .orElseThrow(() -> new RuntimeException("Failed to get host's node spec from node-repo"));
            String hardwareDivergence = maintainer.getHardwareDivergence(node);

            // Only update hardware divergence if there is a change.
            if (!node.getHardwareDivergence().orElse("null").equals(hardwareDivergence)) {
                NodeAttributes nodeAttributes = new NodeAttributes().withHardwareDivergence(hardwareDivergence);
                nodeRepository.updateNodeAttributes(dockerHostHostName, nodeAttributes);
            }
        } catch (RuntimeException e) {
            log.log(Level.WARNING, "Failed to report hardware divergence", e);
        }
    }

    @Override
    public boolean setResumeStateAndCheckIfResumed(State wantedState) {
        synchronized (monitor) {
            if (this.wantedState != wantedState) {
                log.info("Wanted state change: " + this.wantedState + " -> " + wantedState);
                this.wantedState = wantedState;
                signalWorkToBeDone();
            }

            return currentState == wantedState;
        }
    }

    void signalWorkToBeDone() {
        synchronized (monitor) {
            if (! workToDoNow) {
                workToDoNow = true;
                monitor.notifyAll();<commit_msg>Use getNode<commit_after>// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.vespa.hosted.node.admin.nodeadmin;

import com.yahoo.concurrent.ThreadFactoryFactory;
import com.yahoo.concurrent.classlock.ClassLock;
import com.yahoo.concurrent.classlock.ClassLocking;
import com.yahoo.concurrent.classlock.LockInterruptException;
import com.yahoo.log.LogLevel;
import com.yahoo.vespa.hosted.node.admin.configserver.noderepository.NodeSpec;
import com.yahoo.vespa.hosted.node.admin.configserver.noderepository.NodeRepository;
import com.yahoo.vespa.hosted.node.admin.configserver.orchestrator.Orchestrator;
import com.yahoo.vespa.hosted.node.admin.maintenance.StorageMaintainer;
import com.yahoo.vespa.hosted.node.admin.configserver.noderepository.NodeAttributes;
import com.yahoo.vespa.hosted.node.admin.configserver.orchestrator.OrchestratorException;
import com.yahoo.vespa.hosted.node.admin.provider.NodeAdminStateUpdater;
import com.yahoo.vespa.hosted.node.admin.configserver.HttpException;
import com.yahoo.vespa.hosted.provision.Node;

import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Collectors;

import static com.yahoo.vespa.hosted.node.admin.provider.NodeAdminStateUpdater.State.RESUMED;
import static com.yahoo.vespa.hosted.node.admin.provider.NodeAdminStateUpdater.State.SUSPENDED_NODE_ADMIN;
import static com.yahoo.vespa.hosted.node.admin.provider.NodeAdminStateUpdater.State.TRANSITIONING;

/**
 * Pulls information from node repository and forwards containers to run to node admin.
 *
 * @author dybis, stiankri
 */
public class NodeAdminStateUpdaterImpl implements NodeAdminStateUpdater {
    static final Duration FREEZE_CONVERGENCE_TIMEOUT = Duration.ofMinutes(5);

    private final AtomicBoolean terminated = new AtomicBoolean(false);
    private State currentState = SUSPENDED_NODE_ADMIN;
    private State wantedState = RESUMED;
    private boolean workToDoNow = true;

    private final Object monitor = new Object();

    private final Logger log = Logger.getLogger(NodeAdminStateUpdater.class.getName());
    private final ScheduledExecutorService specVerifierScheduler =
            Executors.newScheduledThreadPool(1, ThreadFactoryFactory.getDaemonThreadFactory("specverifier"));
    private final Thread loopThread;

    private final NodeRepository nodeRepository;
    private final Orchestrator orchestrator;
    private final NodeAdmin nodeAdmin;
    private final Clock clock;
    private final String dockerHostHostName;
    private final Duration nodeAdminConvergeStateInterval;

    private final Optional<ClassLocking> classLocking;
    private Optional<ClassLock> classLock = Optional.empty();
    private Instant lastTick;

    public NodeAdminStateUpdaterImpl(
            NodeRepository nodeRepository,
            Orchestrator orchestrator,
            StorageMaintainer storageMaintainer,
            NodeAdmin nodeAdmin,
            String dockerHostHostName,
            Clock clock,
            Duration nodeAdminConvergeStateInterval,
            Optional<ClassLocking> classLocking) {
        log.info(objectToString() + ": Creating object");
        this.nodeRepository = nodeRepository;
        this.orchestrator = orchestrator;
        this.nodeAdmin = nodeAdmin;
        this.dockerHostHostName = dockerHostHostName;
        this.clock = clock;
        this.nodeAdminConvergeStateInterval = nodeAdminConvergeStateInterval;
        this.classLocking = classLocking;
        this.lastTick = clock.instant();

        this.loopThread = new Thread(() -> {
            if (classLocking.isPresent()) {
                log.info(objectToString() + ": Acquiring lock");
                try {
                    classLock = Optional.of(classLocking.get().lockWhile(NodeAdminStateUpdater.class, () -> !terminated.get()));
                } catch (LockInterruptException e) {
                    classLock = Optional.empty();
                    return;
                }
            }

            log.info(objectToString() + ": Starting threads and schedulers");
            nodeAdmin.start();
            specVerifierScheduler.scheduleWithFixedDelay(() ->
                    updateHardwareDivergence(storageMaintainer), 5, 60, TimeUnit.MINUTES);

            while (! terminated.get()) {
                tick();
            }
        });
        this.loopThread.setName("tick-NodeAdminStateUpdater");
    }

    private String objectToString() {
        return this.getClass().getSimpleName() + "@" + Integer.toString(System.identityHashCode(this));
    }

    @Override
    public Map<String, Object> getDebugPage() {
        Map<String, Object> debug = new LinkedHashMap<>();
        synchronized (monitor) {
            debug.put("dockerHostHostName", dockerHostHostName);
            debug.put("wantedState", wantedState);
            debug.put("currentState", currentState);
            debug.put("NodeAdmin", nodeAdmin.debugInfo());
        }
        return debug;
    }

    private void updateHardwareDivergence(StorageMaintainer maintainer) {
        if (currentState != RESUMED) return;

        try {
            NodeSpec node = nodeRepository.getNode(dockerHostHostName);
            String hardwareDivergence = maintainer.getHardwareDivergence(node);

            // Only update hardware divergence if there is a change.
            if (!node.getHardwareDivergence().orElse("null").equals(hardwareDivergence)) {
                NodeAttributes nodeAttributes = new NodeAttributes().withHardwareDivergence(hardwareDivergence);
                nodeRepository.updateNodeAttributes(dockerHostHostName, nodeAttributes);
            }
        } catch (RuntimeException e) {
            log.log(Level.WARNING, "Failed to report hardware divergence", e);
        }
    }

    @Override
    public boolean setResumeStateAndCheckIfResumed(State wantedState) {
        synchronized (monitor) {
            if (this.wantedState != wantedState) {
                log.info("Wanted state change: " + this.wantedState + " -> " + wantedState);
                this.wantedState = wantedState;
                signalWorkToBeDone();
            }

            return currentState == wantedState;
        }
    }

    void signalWorkToBeDone() {
        synchronized (monitor) {
            if (! workToDoNow) {
                workToDoNow = true;
                monitor.notifyAll();
{'fraction_non_alphanumeric': 0.09091568712813815, 'fraction_numerical': 0.0015962850094325932, 'mean_word_length': 3.1266467065868264, 'pattern_counts': {'":': 6, '<': 15, '<?xml version=': 0, '>': 24, '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': '2282640', 'n_tokens_mistral': 3768, 'n_tokens_neox': 3625, 'n_words': 728}