/**
 * Copyright 2016 Yahoo Inc.
 *
 * Licensed 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 com.yahoo.pulsar.broker.loadbalance.impl;

import java.util.ArrayList;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.yahoo.pulsar.broker.BrokerData;
import com.yahoo.pulsar.broker.BundleData;
import com.yahoo.pulsar.broker.LocalBrokerData;
import com.yahoo.pulsar.broker.ServiceConfiguration;
import com.yahoo.pulsar.broker.TimeAverageBrokerData;
import com.yahoo.pulsar.broker.TimeAverageMessageData;
import com.yahoo.pulsar.broker.loadbalance.LoadData;
import com.yahoo.pulsar.broker.loadbalance.ModularLoadManagerStrategy;

/**
 * Placement strategy which selects a broker based on which one has the least long term message rate.
 */
public class LeastLongTermMessageRate implements ModularLoadManagerStrategy {
    private static Logger log = LoggerFactory.getLogger(LeastLongTermMessageRate.class);

    // Maintain this list to reduce object creation.
    private ArrayList<String> bestBrokers;

    public LeastLongTermMessageRate(final ServiceConfiguration conf) {
        bestBrokers = new ArrayList<>();
    }

    // Form a score for a broker using its preallocated bundle data and time average data.
    // This is done by summing all preallocated long-term message rates and adding them to the broker's overall
    // long-term message rate, which is itself the sum of the long-term message rate of every allocated bundle.
    // Once the total long-term message rate is calculated, the score is then weighted by
    // max_usage < overload_threshold ? 1 / (overload_threshold - max_usage): Inf
    // This weight attempts to discourage the placement of bundles on brokers whose system resource usage is high.
    private static double getScore(final BrokerData brokerData, final ServiceConfiguration conf) {
        final double overloadThreshold = conf.getLoadBalancerBrokerOverloadedThresholdPercentage() / 100.0;
        double totalMessageRate = 0;
        for (BundleData bundleData : brokerData.getPreallocatedBundleData().values()) {
            final TimeAverageMessageData longTermData = bundleData.getLongTermData();
            totalMessageRate += longTermData.getMsgRateIn() + longTermData.getMsgRateOut();
        }
        final TimeAverageBrokerData timeAverageData = brokerData.getTimeAverageData();
        final double maxUsage = brokerData.getLocalData().getMaxResourceUsage();
        if (maxUsage > overloadThreshold) {
            log.warn("Broker {} is overloaded: message rate={}, max usage={}",
                    brokerData.getLocalData().getWebServiceUrl(), maxUsage);
            return Double.POSITIVE_INFINITY;
        }
        // 1 / weight is the proportion of load this machine should receive in proportion to a machine with no system
        // resource burden. This attempts to spread out the load in such a way that machines only become overloaded if
        // there is too much load for the system to handle (e.g., all machines are at least nearly overloaded).
        final double weight = 1 / (overloadThreshold - maxUsage);
        final double timeAverageLongTermMessageRate = timeAverageData.getLongTermMsgRateIn()
                + timeAverageData.getLongTermMsgRateOut();
        final double totalMessageRateEstimate = totalMessageRate + timeAverageLongTermMessageRate;
        final double score = weight * totalMessageRateEstimate;
        log.info("Broker {} has long term message rate {}, weight {}, and score of {}",
                brokerData.getLocalData().getWebServiceUrl(), totalMessageRateEstimate, weight, score);
        return score;
    }

    /**
     * Find a suitable broker to assign the given bundle to.
     * 
     * @param candidates
     *            The candidates for which the bundle may be assigned.
     * @param bundleToAssign
     *            The data for the bundle to assign.
     * @param loadData
     *            The load data from the leader broker.
     * @param conf
     *            The service configuration.
     * @return The name of the selected broker as it appears on ZooKeeper.
     */
    @Override
    public String selectBroker(final Set<String> candidates, final BundleData bundleToAssign, final LoadData loadData,
            final ServiceConfiguration conf) {
        bestBrokers.clear();
        double minScore = Double.POSITIVE_INFINITY;
        // Maintain of list of all the best scoring brokers and then randomly
        // select one of them at the end.
        for (String broker : candidates) {
            final BrokerData brokerData = loadData.getBrokerData().get(broker);
            final double score = getScore(brokerData, conf);
            if (score == Double.POSITIVE_INFINITY) {
                final LocalBrokerData localData = brokerData.getLocalData();
                log.warn(
                        "Broker {} is overloaded: CPU: {}%, MEMORY: {}%, DIRECT MEMORY: {}%, BANDWIDTH IN: {}%, "
                                + "BANDWIDTH OUT: {}%",
                        broker, localData.getCpu().percentUsage(), localData.getMemory().percentUsage(),
                        localData.getDirectMemory().percentUsage(), localData.getBandwidthIn().percentUsage(),
                        localData.getBandwidthOut().percentUsage());

            }
            if (score < minScore) {
                // Clear best brokers since this score beats the other brokers.
                bestBrokers.clear();
                bestBrokers.add(broker);
                minScore = score;
            } else if (score == minScore) {
                // Add this broker to best brokers since it ties with the best score.
                bestBrokers.add(broker);
            }
        }
        if (bestBrokers.isEmpty()) {
            // All brokers are overloaded.
            // Assign randomly in this case.
            bestBrokers.addAll(candidates);
        }
        return bestBrokers.get(ThreadLocalRandom.current().nextInt(bestBrokers.size()));
    }
}

@@ -2036,144 +2036,45 @@
  // 
-Once the total long-term message rate is calculated, the score is then weighted by%0A    // max_usage %3C overload_threshold ? 1 / (
+Any broker at (or above) the 
 overload
 _thr
@@ -2069,17 +2069,17 @@
 overload
-_
+ 
 threshol
@@ -2084,139 +2084,46 @@
 old 
-- max_usage): Inf%0A    // This weight attempts to discourage the placement of bundles on brokers whose system resource usage is high
+will have a score of POSITIVE_INFINITY
 .%0A  
@@ -2904,58 +2904,21 @@
 d: m
-essage rate=%7B%7D, max usage=%7B%7D%22,%0A                   
+ax usage=%7B%7D%22,
  bro
@@ -3030,423 +3030,8 @@
   %7D%0A
-        // 1 / weight is the proportion of load this machine should receive in proportion to a machine with no system%0A        // resource burden. This attempts to spread out the load in such a way that machines only become overloaded if%0A        // there is too much load for the system to handle (e.g., all machines are at least nearly overloaded).%0A        final double weight = 1 / (overloadThreshold - maxUsage);%0A
     
@@ -3277,24 +3277,25 @@
 geRate;%0A
+%0A
         
 final do
@@ -3290,64 +3290,40 @@
     
-final double score = weight * totalMessageRateEstimate;%0A
+if (log.isDebugEnabled()) %7B%0A    
     
@@ -3334,12 +3334,13 @@
 log.
-info
+debug
 (%22Br
@@ -3380,39 +3380,15 @@
 e %7B%7D
-, weight %7B%7D, and score of %7B%7D%22,%0A
+%22,%0A    
     
@@ -3473,45 +3473,59 @@
 mate
-, weight, score);%0A        return scor
+);%0A        %7D%0A        return totalMessageRateEstimat
 e;%0A 

Remove broker weights for ModularLoadManager (#442)