<commit_before>/*
 Licensed to <NAME>. under one
 or more contributor license agreements. See the NOTICE file
 distributed with this work for additional information
 regarding copyright ownership. <NAME>.l. licenses this file
 to you under the Apache License, Version 2.0 (the
 "License"); you may not use this file except in compliance
 with the License.  You may obtain a copy of the License at

 http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing,
 software distributed under the License is distributed on an
 "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 KIND, either express or implied.  See the License for the
 specific language governing permissions and limitations
 under the License.

 */
package majordodo.task;

import majordodo.clientfacade.TaskStatusView;
import majordodo.clientfacade.WorkerStatusView;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Collectors;
import majordodo.clientfacade.CodePoolView;
import majordodo.clientfacade.TransactionStatus;
import majordodo.codepools.CodePool;

/**
 * Replicated status of the broker. Each broker, leader or follower, contains a
 * copy of this status. The status is replicated to follower by reading the
 * StatusChangesLog
 *
 * @author enrico.olivelli
 */
public class BrokerStatus {

    private static final Logger LOGGER = Logger.getLogger(BrokerStatus.class.getName());

    private final Map<Long, Task> tasks = new HashMap<>();
    private final Map<Long, Transaction> transactions = new HashMap<>();

    private final Map<String, WorkerStatus> workers = new HashMap<>();
    private final Map<String, CodePool> codePools = new HashMap<>();
    private final AtomicLong newTaskId = new AtomicLong();
    private final AtomicLong newTransactionId = new AtomicLong();
    private long maxTaskId = -1;
    private long maxTransactionId = -1;
    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    private final StatusChangesLog log;
    private LogSequenceNumber lastLogSequenceNumber;
    private final AtomicInteger checkpointsCount = new AtomicInteger();
    private final SlotsManager slotsManager = new SlotsManager();
    private final BrokerStatusStats stats = new BrokerStatusStats();

    public WorkerStatus getWorkerStatus(String workerId) {
        return workers.get(workerId);
    }

    public BrokerStatus(StatusChangesLog log) {
        this.log = log;
    }

    public Map<String, Long> getActualSlots() {
        return slotsManager.getActualSlots();
    }

    public List<TransactionStatus> getAllTransactions() {
        List<TransactionStatus> result = new ArrayList<>();
        lock.readLock().lock();
        try {
            transactions.values().stream().forEach((k) -> {
                result.add(createTransactionStatusView(k));
            });
        } finally {
            lock.readLock().unlock();
        }
        return result;
    }

    public List<WorkerStatusView> getAllWorkers() {
        List<WorkerStatusView> result = new ArrayList<>();
        lock.readLock().lock();
        try {
            workers.values().stream().forEach((k) -> {
                result.add(createWorkerStatusView(k));
            });
        } finally {
            lock.readLock().unlock();
        }
        return result;
    }

    public List<TaskStatusView> getAllTasks() {
        List<TaskStatusView> result = new ArrayList<>();
        lock.readLock().lock();
        try {
            tasks.values().stream().forEach((k) -> {
                result.add(createTaskStatusView(k));
            });
        } finally {
            lock.readLock().unlock();
        }
        return result;
    }

    private TaskStatusView createTaskStatusView(Task task) {
        if (task == null) {
            return null;
        }
        TaskStatusView s = new TaskStatusView();
        s.setCreatedTimestamp(task.getCreatedTimestamp());
        s.setUser(task.getUserId());
        s.setWorkerId(task.getWorkerId());
        s.setStatus(task.getStatus());
        s.setTaskId(task.getTaskId());
        s.setData(task.getParameter());
        s.setType(task.getType());
        s.setResult(task.getResult());
        s.setAttempts(task.getAttempts());
        s.setMaxattempts(task.getMaxattempts());
        s.setSlot(task.getSlot());
        s.setExecutionDeadline(task.getExecutionDeadline());
        if (task.getMode() != null && !Task.MODE_EXECUTE_FACTORY.equals(task.getMode())) {
            s.setMode(task.getMode());
        }
        if (task.getCodepool() != null) {
            s.setCodePoolId(task.getCodepool());
        }
        return s;
    }

    private WorkerStatusView createWorkerStatusView(WorkerStatus k) {
        WorkerStatusView res = new WorkerStatusView();
        res.setId(k.getWorkerId());
        res.setLocation(k.getWorkerLocation());
        res.setLastConnectionTs(k.getLastConnectionTs());
        String s;
        switch (k.getStatus()) {
            case WorkerStatus.STATUS_CONNECTED:
                s = "CONNECTED";
                break;
            case WorkerStatus.STATUS_DEAD:
                s = "DEAD";
                break;
            case WorkerStatus.STATUS_DISCONNECTED:
                s = "DISCONNECTED";
                break;
            default:
                s = "?" + k.getStatus();
        }
        res.setProcessId(k.getProcessId());
        res.setStatus(s);
        return res;
    }

    public Collection<WorkerStatus> getWorkersAtBoot() {
        return workers.values();
    }

    public Collection<Task> getTasksAtBoot() {
        return tasks.values();
    }

    public Collection<Transaction> getTransactionsAtBoot() {
        return transactions.values();
    }

    public long nextTaskId() {
        return newTaskId.incrementAndGet();
    }

    public long nextTransactionId() {
        return newTransactionId.incrementAndGet();
    }

    public int getCheckpointsCount() {
        return checkpointsCount.get();
    }

    private void purgeAbandonedCodePools() throws LogNotAvailableException {
        List<String> toPurge = new ArrayList<>();
        lock.readLock().lock();
        try {
            for (CodePool codePool : codePools.values()) {
                if (codePool.getTtl() > 0) {
                    long delta = System.currentTimeMillis() - codePool.getCreationTimestamp();
                    if (delta > codePool.getTtl()) {
                        String codePoolId = codePool.getId();
                        boolean hasTasks = false;
                        for (Task t : tasks.values()) {
                            if (codePoolId.equals(t.getCodepool())) {
                                switch (t.getStatus()) {
                                    case Task.STATUS_RUNNING:
                                    case Task.STATUS_WAITING:
                                        hasTasks = true;                                        
                                        break;
                                }
                                if (hasTasks) {
                                    break;
                                }
                            }
                        }
                        if (!hasTasks) {
                            toPurge.add(codePoolId);
                        }
                    }
                    
                }
            }
        } finally {
            lock.readLock().unlock();
        }
        for (String codePoolId : toPurge) {
            StatusEdit rollback = StatusEdit.DELETE_CODEPOOL(codePoolId);
            LOGGER.log(Level.SEVERE, "Detected abandoned codepool {0}, dropping", new Object[]{codePoolId});
            applyModification(rollback);
        }
    }

    private void purgeAbandonedTransactions(long ttl) throws LogNotAvailableException {
        List<Long> staleTransactions = new ArrayList<>();
        lock.readLock().lock();
        try {
            long deadLine = System.currentTimeMillis() - ttl;
            transactions.values().stream().filter((t) -> (t.getCreationTimestamp() < deadLine)).forEach((t) -> {
                staleTransactions.add(t.getTransactionId());
            });
        } finally {
            lock.readLock().unlock();
        }
        for (long transactionId : staleTransactions) {
            StatusEdit rollback = StatusEdit.ROLLBACK_TRANSACTION(transactionId);
            LOGGER.log(Level.SEVERE, "Detected abandoned transaction {0}, Forcing rollback", new Object[]{transactionId});
            applyModification(rollback);
        }
    }

    public void checkpoint(long transactionsTtl) throws LogNotAvailableException {
        checkpointsCount.incrementAndGet();
        LOGGER.info("checkpoint");

        if (transactionsTtl > 0) {
            purgeAbandonedTransactions(transactionsTtl);
        }
        purgeAbandonedCodePools();

        BrokerStatusSnapshot snapshot = createSnapshot();

        this.log.checkpoint(snapshot);
    }

    public BrokerStatusSnapshot createSnapshot() {
        lock.readLock().lock();
        try {
            BrokerStatusSnapshot snap = new BrokerStatusSnapshot(maxTaskId, maxTransactionId, lastLogSequenceNumber);
            for (Task task : tasks.values()) {
                snap.tasks.add(task.cloneForSnapshot());
            }
            for (WorkerStatus status : workers.values()) {
                snap.workers.add(status.cloneForSnapshot());
            }
            for (Transaction status : transactions.values()) {
                snap.transactions.add(status.cloneForSnapshot());
            }
            return snap;
        } finally {
            lock.readLock().unlock();
        }
    }

    public void close() {
        try {
            this.log.close();
        } catch (LogNotAvailableException sorry) {
            LOGGER.log(Level.SEVERE, "Error while closing transaction log", sorry);
        }

    }

    public Set<Long> purgeFinishedTasksAndSignalExpiredTasks(int finishedTasksRetention, int maxExpiredPerCycle) {
        long now = System.currentTimeMillis();
        long finished_deadline = now - finishedTasksRetention;

        Set<Long> expired = new HashSet<>();
        int expiredcount = 0;
        this.lock.writeLock().lock();
        try {
            // tasks are only purged from memry, not from logs
            // in case of broker restart it may re-appear
            for (Iterator<Map.Entry<Long, Task>> it = tasks.entrySet().iterator(); it.hasNext();) {
                Map.Entry<Long, Task> taskEntry = it.next();
                Task t = taskEntry.getValue();
                switch (t.getStatus()) {
                    case Task.STATUS_WAITING:
                        if (expiredcount < maxExpiredPerCycle) {
                            long taskdeadline = t.getExecutionDeadline();
                            if (taskdeadline > 0 && taskdeadline < now) {
                                expired.add(t.getTaskId());
                                expiredcount++;
                                LOGGER.log(Level.INFO, "task {0}, created at {1}, expired, deadline {2}", new Object[]{t.getTaskId(), new java.util.Date(t.getCreatedTimestamp()), new java.util.Date(taskdeadline)});
                            }
                        }
                        break;
                    case Task.STATUS_ERROR:
                    case Task.STATUS_FINISHED:
                        if (t.getCreatedTimestamp() < finished_deadline) {
                            LOGGER.log(Level.INFO, "purging finished task {0} slot {2}, created at {1}", new Object[]{t.getTaskId(), new java.util.Date(t.getCreatedTimestamp()), t.getSlot()});
                            it.remove();
                            stats.taskStatusChange(t.getStatus(), -1);
                        }
                        break;
                }
            }
        } finally {
            this.lock.writeLock().unlock();
        }
        return expired;
    }

    public void followTheLeader() throws InterruptedException {<commit_msg>MAJ-58<commit_after>/*
 Licensed to <NAME>. under one
 or more contributor license agreements. See the NOTICE file
 distributed with this work for additional information
 regarding copyright ownership. <NAME>. licenses this file
 to you under the Apache License, Version 2.0 (the
 "License"); you may not use this file except in compliance
 with the License.  You may obtain a copy of the License at

 http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing,
 software distributed under the License is distributed on an
 "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 KIND, either express or implied.  See the License for the
 specific language governing permissions and limitations
 under the License.

 */
package majordodo.task;

import majordodo.clientfacade.TaskStatusView;
import majordodo.clientfacade.WorkerStatusView;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Collectors;
import majordodo.clientfacade.CodePoolView;
import majordodo.clientfacade.TransactionStatus;
import majordodo.codepools.CodePool;

/**
 * Replicated status of the broker. Each broker, leader or follower, contains a
 * copy of this status. The status is replicated to follower by reading the
 * StatusChangesLog
 *
 * @author enrico.olivelli
 */
public class BrokerStatus {

    private static final Logger LOGGER = Logger.getLogger(BrokerStatus.class.getName());

    private final Map<Long, Task> tasks = new HashMap<>();
    private final Map<Long, Transaction> transactions = new HashMap<>();

    private final Map<String, WorkerStatus> workers = new HashMap<>();
    private final Map<String, CodePool> codePools = new HashMap<>();
    private final AtomicLong newTaskId = new AtomicLong();
    private final AtomicLong newTransactionId = new AtomicLong();
    private long maxTaskId = -1;
    private long maxTransactionId = -1;
    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    private final StatusChangesLog log;
    private LogSequenceNumber lastLogSequenceNumber;
    private final AtomicInteger checkpointsCount = new AtomicInteger();
    private final SlotsManager slotsManager = new SlotsManager();
    private final BrokerStatusStats stats = new BrokerStatusStats();

    public WorkerStatus getWorkerStatus(String workerId) {
        return workers.get(workerId);
    }

    public BrokerStatus(StatusChangesLog log) {
        this.log = log;
    }

    public Map<String, Long> getActualSlots() {
        return slotsManager.getActualSlots();
    }

    public List<TransactionStatus> getAllTransactions() {
        List<TransactionStatus> result = new ArrayList<>();
        lock.readLock().lock();
        try {
            transactions.values().stream().forEach((k) -> {
                result.add(createTransactionStatusView(k));
            });
        } finally {
            lock.readLock().unlock();
        }
        return result;
    }

    public List<WorkerStatusView> getAllWorkers() {
        List<WorkerStatusView> result = new ArrayList<>();
        lock.readLock().lock();
        try {
            workers.values().stream().forEach((k) -> {
                result.add(createWorkerStatusView(k));
            });
        } finally {
            lock.readLock().unlock();
        }
        return result;
    }

    public List<TaskStatusView> getAllTasks() {
        List<TaskStatusView> result = new ArrayList<>();
        lock.readLock().lock();
        try {
            tasks.values().stream().forEach((k) -> {
                result.add(createTaskStatusView(k));
            });
        } finally {
            lock.readLock().unlock();
        }
        return result;
    }

    private TaskStatusView createTaskStatusView(Task task) {
        if (task == null) {
            return null;
        }
        TaskStatusView s = new TaskStatusView();
        s.setCreatedTimestamp(task.getCreatedTimestamp());
        s.setUser(task.getUserId());
        s.setWorkerId(task.getWorkerId());
        s.setStatus(task.getStatus());
        s.setTaskId(task.getTaskId());
        s.setData(task.getParameter());
        s.setType(task.getType());
        s.setResult(task.getResult());
        s.setAttempts(task.getAttempts());
        s.setMaxattempts(task.getMaxattempts());
        s.setSlot(task.getSlot());
        s.setExecutionDeadline(task.getExecutionDeadline());
        if (task.getMode() != null && !Task.MODE_EXECUTE_FACTORY.equals(task.getMode())) {
            s.setMode(task.getMode());
        }
        if (task.getCodepool() != null) {
            s.setCodePoolId(task.getCodepool());
        }
        return s;
    }

    private WorkerStatusView createWorkerStatusView(WorkerStatus k) {
        WorkerStatusView res = new WorkerStatusView();
        res.setId(k.getWorkerId());
        res.setLocation(k.getWorkerLocation());
        res.setLastConnectionTs(k.getLastConnectionTs());
        String s;
        switch (k.getStatus()) {
            case WorkerStatus.STATUS_CONNECTED:
                s = "CONNECTED";
                break;
            case WorkerStatus.STATUS_DEAD:
                s = "DEAD";
                break;
            case WorkerStatus.STATUS_DISCONNECTED:
                s = "DISCONNECTED";
                break;
            default:
                s = "?" + k.getStatus();
        }
        res.setProcessId(k.getProcessId());
        res.setStatus(s);
        return res;
    }

    public Collection<WorkerStatus> getWorkersAtBoot() {
        return workers.values();
    }

    public Collection<Task> getTasksAtBoot() {
        return tasks.values();
    }

    public Collection<Transaction> getTransactionsAtBoot() {
        return transactions.values();
    }

    public long nextTaskId() {
        return newTaskId.incrementAndGet();
    }

    public long nextTransactionId() {
        return newTransactionId.incrementAndGet();
    }

    public int getCheckpointsCount() {
        return checkpointsCount.get();
    }

    private void purgeAbandonedCodePools() throws LogNotAvailableException {
        List<String> toPurge = new ArrayList<>();
        lock.readLock().lock();
        try {
            for (CodePool codePool : codePools.values()) {
                if (codePool.getTtl() > 0) {
                    long delta = System.currentTimeMillis() - codePool.getCreationTimestamp();
                    if (delta > codePool.getTtl()) {
                        String codePoolId = codePool.getId();
                        boolean hasTasks = false;
                        for (Task t : tasks.values()) {
                            if (codePoolId.equals(t.getCodepool())) {
                                switch (t.getStatus()) {
                                    case Task.STATUS_RUNNING:
                                    case Task.STATUS_WAITING:
                                        hasTasks = true;                                        
                                        break;
                                }
                                if (hasTasks) {
                                    break;
                                }
                            }
                        }
                        if (!hasTasks) {
                            toPurge.add(codePoolId);
                        }
                    }
                    
                }
            }
        } finally {
            lock.readLock().unlock();
        }
        for (String codePoolId : toPurge) {
            StatusEdit rollback = StatusEdit.DELETE_CODEPOOL(codePoolId);
            LOGGER.log(Level.SEVERE, "Detected abandoned codepool {0}, dropping", new Object[]{codePoolId});
            applyModification(rollback);
        }
    }

    private void purgeAbandonedTransactions(long ttl) throws LogNotAvailableException {
        List<Long> staleTransactions = new ArrayList<>();
        lock.readLock().lock();
        try {
            long deadLine = System.currentTimeMillis() - ttl;
            transactions.values().stream().filter((t) -> (t.getCreationTimestamp() < deadLine)).forEach((t) -> {
                staleTransactions.add(t.getTransactionId());
            });
        } finally {
            lock.readLock().unlock();
        }
        for (long transactionId : staleTransactions) {
            StatusEdit rollback = StatusEdit.ROLLBACK_TRANSACTION(transactionId);
            LOGGER.log(Level.SEVERE, "Detected abandoned transaction {0}, Forcing rollback", new Object[]{transactionId});
            applyModification(rollback);
        }
    }

    public void checkpoint(long transactionsTtl) throws LogNotAvailableException {
        checkpointsCount.incrementAndGet();
        LOGGER.info("checkpoint");

        if (transactionsTtl > 0) {
            purgeAbandonedTransactions(transactionsTtl);
        }
        purgeAbandonedCodePools();

        BrokerStatusSnapshot snapshot = createSnapshot();

        this.log.checkpoint(snapshot);
    }

    public BrokerStatusSnapshot createSnapshot() {
        lock.readLock().lock();
        try {
            BrokerStatusSnapshot snap = new BrokerStatusSnapshot(maxTaskId, maxTransactionId, lastLogSequenceNumber);
            for (Task task : tasks.values()) {
                snap.tasks.add(task.cloneForSnapshot());
            }
            for (WorkerStatus status : workers.values()) {
                snap.workers.add(status.cloneForSnapshot());
            }
            for (Transaction status : transactions.values()) {
                snap.transactions.add(status.cloneForSnapshot());
            }
            return snap;
        } finally {
            lock.readLock().unlock();
        }
    }

    public void close() {
        try {
            this.log.close();
        } catch (LogNotAvailableException sorry) {
            LOGGER.log(Level.SEVERE, "Error while closing transaction log", sorry);
        }

    }

    public Set<Long> purgeFinishedTasksAndSignalExpiredTasks(int finishedTasksRetention, int maxExpiredPerCycle) {
        long now = System.currentTimeMillis();
        long finished_deadline = now - finishedTasksRetention;

        Set<Long> expired = new HashSet<>();
        int expiredcount = 0;
        this.lock.writeLock().lock();
        // when running in FOLLOWER MODE we cannot expire tasks, but we need to remove them from memory, see MAJ-58
        boolean allowExpire = this.log.isLeader() && this.log.isWritable();
        try {
            // tasks are only purged from memry, not from logs
            // in case of broker restart it may re-appear
            for (Iterator<Map.Entry<Long, Task>> it = tasks.entrySet().iterator(); it.hasNext();) {
                Map.Entry<Long, Task> taskEntry = it.next();
                Task t = taskEntry.getValue();
                switch (t.getStatus()) {
                    case Task.STATUS_WAITING:
                        if (expiredcount < maxExpiredPerCycle && allowExpire) {
                            long taskdeadline = t.getExecutionDeadline();
                            if (taskdeadline > 0 && taskdeadline < now) {
                                expired.add(t.getTaskId());
                                expiredcount++;
                                LOGGER.log(Level.INFO, "task {0}, created at {1}, expired, deadline {2}", new Object[]{t.getTaskId(), new java.util.Date(t.getCreatedTimestamp()), new java.util.Date(taskdeadline)});
                            }
                        }
                        break;
                    case Task.STATUS_ERROR:
                    case Task.STATUS_FINISHED:
                        if (t.getCreatedTimestamp() < finished_deadline) {
                            LOGGER.log(Level.INFO, "purging finished task {0} slot {2}, created at {1}", new Object[]{t.getTaskId(), new java.util.Date(t.getCreatedTimestamp()), t.getSlot()});
                            it.remove();
                            stats.taskStatusChange(t.getStatus(), -1);
                        }
                        break;
                }
            }
        } finally {
            this.lock.writeLock().unlock();
        }
        return expired;
    }

    public void followTheLeader() throws InterruptedException {
{'fraction_non_alphanumeric': 0.10264874323894368, 'fraction_numerical': 0.0016703786191536749, 'mean_word_length': 1.9464494961331147, 'pattern_counts': {'":': 0, '<': 77, '<?xml version=': 0, '>': 87, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '530671', 'n_tokens_mistral': 6516, 'n_tokens_neox': 6323, 'n_words': 1469}