<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with this
 * work for additional information regarding copyright ownership. The ASF
 * 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 io.tilt.minka.core.monitor;

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

import java.time.Instant;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

import org.apache.commons.lang.Validate;
import org.joda.time.DateTime;
import org.json.JSONArray;
import org.json.JSONObject;

import io.tilt.minka.api.Config;
import io.tilt.minka.core.leader.balancer.Balancer.BalancerMetadata;
import io.tilt.minka.core.leader.data.CommitedState;
import io.tilt.minka.core.leader.data.Scheme;
import io.tilt.minka.core.leader.data.UncommitedChanges;
import io.tilt.minka.core.leader.distributor.ChangePlan;
import io.tilt.minka.core.task.LeaderAware;
import io.tilt.minka.domain.EntityEvent;
import io.tilt.minka.domain.EntityState;
import io.tilt.minka.domain.ShardEntity;
import io.tilt.minka.domain.ShardedPartition;
import io.tilt.minka.shard.Shard;
import io.tilt.minka.utils.CollectionUtils;
import io.tilt.minka.utils.CollectionUtils.SlidingSortedSet;

/**
 * Read only views built at request about the system state  
 * JSON format.
 * 
 * @author <NAME>
 * @since Nov 6, 2016
 */
public class LeaderMonitor {

	private final LeaderAware leaderAware;
	private final Config config;
	
	private final Scheme scheme;
	private final ShardedPartition partition;
	
	private final ChangePlan[] lastPlan = {null};
	private final SlidingSortedSet<String> changePlanHistory;
	private final SlidingSortedSet<Long> planids;

	public LeaderMonitor(
			final LeaderAware leaderAware,
			final Config config,
			final Scheme scheme,
			final ShardedPartition partition) {
		super();
		
		this.leaderAware = requireNonNull(leaderAware); 
		this.config = requireNonNull(config);
		this.scheme = requireNonNull(scheme);
		this.partition = requireNonNull(partition);
		
		this.changePlanHistory = CollectionUtils.sliding(20);
		this.planids = CollectionUtils.sliding(10);
		
		scheme.addChangeObserver(()->{
			try {
				if (lastPlan[0]!=null) {
					if (!lastPlan[0].equals(scheme.getCurrentPlan())) {
						changePlanHistory.add(SystemStateMonitor.toJson(lastPlan[0]));
						planids.add(lastPlan[0].getId());
					}
				}
				lastPlan[0] = scheme.getCurrentPlan();
			} catch (Exception e) {
			}
		});
	}


	/**
	 * <p>
	 * Gives detailed information on the shards, members of the cluster within the namespace, reporting to the leader.
	 * <p>
	 * Full when the current server is the Leader, self shard info. when the server is a Follower.
	 * @return			a String in json format
	 */
	public String shardsToJson()  {
		return SystemStateMonitor.toJson(buildShards(scheme));
	}
	
	/**
	 * <p>
	 * Gives detailed information on change plans built in distributor phase
	 * to accomodate duties into shards. Including any on-going plan, and historic plans.
	 * <p>
	 * Non-Empty only when the current server is the Leader.
	 * @return			a String in json format
	 */
	public String plansToJson() {
		return SystemStateMonitor.toJson(buildPlans().toMap());
	}

	/**
	 * <p>
	 * Gives detailed information on distributed duties and their shard locations
	 * <p>
	 * Non-Empty only when the current server is the Leader.
	 * @return			a String in json format
	 */
	public String distributionToJson() {
		return SystemStateMonitor.toJson(buildDistribution());
	}

	public String palletsToJson() {
		return SystemStateMonitor.toJson(buildPallets());
	}

	/**
	 * 
	 * <p>
	 * Non-Empty only when the current server is the Leader.
	 * @return			a String in json format
	 */
	public String schemeToJson(final boolean detail) {
		Map<String, Object> m = new LinkedHashMap<>(2);
		m.put("commited", buildCommitedState(detail));
		m.put("replicas", buildReplicas(detail));
		m.put("uncommited", buildUncommitedChanges(detail, scheme.getUncommited()));
		return SystemStateMonitor.toJson(m);
	}
	
	private JSONObject buildPlans() {
		final JSONObject js = new JSONObject();
		try {
			final JSONArray arr = new JSONArray();
			js.put("size", changePlanHistory.size());
			changePlanHistory.values().forEach(s-> arr.put(new JSONObject(s)));
			js.put("ids", this.planids.values());
			if (lastPlan[0]!=null && !lastPlan[0].getResult().isClosed()) {
				js.put("current", new JSONObject(SystemStateMonitor.toJson(lastPlan[0])));
			} else if (lastPlan[0]!=null) {
				arr.put(new JSONObject(SystemStateMonitor.toJson(lastPlan[0])));
			}
			
			js.put("history", arr);
		} catch (Exception e) {
		}
		return js;
	}

	private Map<String, Object> buildShards(final Scheme scheme) {
		final Map<String, Object> map = new LinkedHashMap<>(5);
		map.put("namespace", config.getBootstrap().getNamespace());<commit_msg>Monitor: fixes to plans json<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with this
 * work for additional information regarding copyright ownership. The ASF
 * 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 io.tilt.minka.core.monitor;

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

import java.time.Instant;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

import org.apache.commons.lang.Validate;
import org.joda.time.DateTime;
import org.json.JSONArray;
import org.json.JSONObject;

import io.tilt.minka.api.Config;
import io.tilt.minka.core.leader.balancer.Balancer.BalancerMetadata;
import io.tilt.minka.core.leader.data.CommitedState;
import io.tilt.minka.core.leader.data.Scheme;
import io.tilt.minka.core.leader.data.UncommitedChanges;
import io.tilt.minka.core.leader.distributor.ChangePlan;
import io.tilt.minka.core.task.LeaderAware;
import io.tilt.minka.domain.EntityEvent;
import io.tilt.minka.domain.EntityState;
import io.tilt.minka.domain.ShardEntity;
import io.tilt.minka.domain.ShardedPartition;
import io.tilt.minka.shard.Shard;
import io.tilt.minka.utils.CollectionUtils;
import io.tilt.minka.utils.CollectionUtils.SlidingSortedSet;

/**
 * Read only views built at request about the system state  
 * JSON format.
 * 
 * @author <NAME>
 * @since Nov 6, 2016
 */
public class LeaderMonitor {

	private final LeaderAware leaderAware;
	private final Config config;
	
	private final Scheme scheme;
	private final ShardedPartition partition;
	
	private final ChangePlan[] lastPlan = {null};
	private final SlidingSortedSet<String> changePlanHistory;
	private final SlidingSortedSet<Long> planids;

	public LeaderMonitor(
			final LeaderAware leaderAware,
			final Config config,
			final Scheme scheme,
			final ShardedPartition partition) {
		super();
		
		this.leaderAware = requireNonNull(leaderAware); 
		this.config = requireNonNull(config);
		this.scheme = requireNonNull(scheme);
		this.partition = requireNonNull(partition);
		
		this.changePlanHistory = CollectionUtils.sliding(20);
		this.planids = CollectionUtils.sliding(10);
		
		scheme.addChangeObserver(()-> {
			try {
				if (lastPlan[0]!=null) {
					if (!lastPlan[0].equals(scheme.getCurrentPlan())) {
						changePlanHistory.add(SystemStateMonitor.toJson(lastPlan[0]));
						planids.add(lastPlan[0].getId());
					}
				}
				lastPlan[0] = scheme.getCurrentPlan();
			} catch (Exception e) {
			}
		});
	}


	/**
	 * <p>
	 * Gives detailed information on the shards, members of the cluster within the namespace, reporting to the leader.
	 * <p>
	 * Full when the current server is the Leader, self shard info. when the server is a Follower.
	 * @return			a String in json format
	 */
	public String shardsToJson()  {
		return SystemStateMonitor.toJson(buildShards(scheme));
	}
	
	/**
	 * <p>
	 * Gives detailed information on change plans built in distributor phase
	 * to accomodate duties into shards. Including any on-going plan, and historic plans.
	 * <p>
	 * Non-Empty only when the current server is the Leader.
	 * @return			a String in json format
	 */
	public String plansToJson() {
		return SystemStateMonitor.toJson(buildPlans().toMap());
	}

	/**
	 * <p>
	 * Gives detailed information on distributed duties and their shard locations
	 * <p>
	 * Non-Empty only when the current server is the Leader.
	 * @return			a String in json format
	 */
	public String distributionToJson() {
		return SystemStateMonitor.toJson(buildDistribution());
	}

	public String palletsToJson() {
		return SystemStateMonitor.toJson(buildPallets());
	}

	/**
	 * 
	 * <p>
	 * Non-Empty only when the current server is the Leader.
	 * @return			a String in json format
	 */
	public String schemeToJson(final boolean detail) {
		Map<String, Object> m = new LinkedHashMap<>(2);
		m.put("commited", buildCommitedState(detail));
		m.put("replicas", buildReplicas(detail));
		m.put("uncommited", buildUncommitedChanges(detail, scheme.getUncommited()));
		return SystemStateMonitor.toJson(m);
	}
	
	private JSONObject buildPlans() {
		final JSONObject js = new JSONObject();
		try {
			js.put("size", changePlanHistory.size() + (lastPlan[0]!=null ? 1 : 0));
			js.put("ids", this.planids.values());
			if (lastPlan[0]!=null) {
				js.put("last", new JSONObject(SystemStateMonitor.toJson(lastPlan[0])));
			}
			
			final JSONArray arr = new JSONArray();
			final Iterator<String> it = changePlanHistory.descend();
			while(it.hasNext()) {
				final String s = it.next();
				arr.put(new JSONObject(s));
			}
			js.put("closed", arr);
			
		} catch (Exception e) {
		}
		return js;
	}

	private Map<String, Object> buildShards(final Scheme scheme) {
		final Map<String, Object> map = new LinkedHashMap<>(5);
		map.put("namespace", config.getBootstrap().getNamespace());
{'fraction_non_alphanumeric': 0.11181228335259089, 'fraction_numerical': 0.0044440494178295264, 'mean_word_length': 4.287593984962406, 'pattern_counts': {'":': 0, '<': 34, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '3109513', 'n_tokens_mistral': 3735, 'n_tokens_neox': 3412, 'n_words': 972}