/**
 * Copyright 2016 Pinterest, 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.pinterest.teletraan.worker;

import com.pinterest.arcee.autoscaling.AutoScaleGroupManager;
import com.pinterest.arcee.bean.AutoScalingGroupBean;
import com.pinterest.arcee.metrics.MetricSource;
import com.pinterest.deployservice.ServiceContext;
import com.pinterest.arcee.dao.GroupInfoDAO;
import com.pinterest.deployservice.dao.HostDAO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.util.List;
import java.util.Collections;
import java.util.HashMap;


public class GroupInfoUpdater implements Runnable {
    private static final Logger LOG = LoggerFactory.getLogger(GroupInfoUpdater.class);
    private static final String COUNTER_NAME = "autoscaling.%s.size";
    private GroupInfoDAO groupInfoDAO;
    private HostDAO hostDAO;
    private HashMap<String, String> tags;
    private MetricSource metricSource;
    private AutoScaleGroupManager autoScaleGroupManager;

    public GroupInfoUpdater(ServiceContext context) {
        groupInfoDAO = context.getGroupInfoDAO();
        hostDAO = context.getHostDAO();
        metricSource = context.getMetricSource();
        autoScaleGroupManager = context.getAutoScaleGroupManager();
        tags = new HashMap<>();
        try {
            tags.put("host", InetAddress.getLocalHost().getHostName());
        } catch (Exception t) {
            LOG.error("Failed to get local host name");
        }
    }

    private void sendGroupMetrics(String groupName) throws Exception {
        Long groupSize = hostDAO.getGroupSize(groupName);
        AutoScalingGroupBean autoScalingGroupBean = autoScaleGroupManager.getAutoScalingGroupInfoByName(String.format("%s-spot", groupName));
        Long currentTime = System.currentTimeMillis();
        if (autoScalingGroupBean != null) {
            Integer spotInstanceCount = autoScalingGroupBean.getInstances().size();
            groupSize = groupSize - spotInstanceCount;
            metricSource.export(String.format(COUNTER_NAME, String.format("%s-spot", groupName)), tags, spotInstanceCount.doubleValue(), currentTime);
        }
        metricSource.export(String.format(COUNTER_NAME, groupName), tags, groupSize.doubleValue(), currentTime);
    }

    public void processBatch() throws Exception {
        List<String> groups = groupInfoDAO.getExistingGroups(1, 1000);
        Collections.shuffle(groups);
        for (String group : groups) {
            try {
                LOG.info("Start to send metrics to tsd for group: {}", group);
                sendGroupMetrics(group);
                // sleep 500 milliseconds for the next operation
                Thread.sleep(500);
            } catch (Exception ex) {
                LOG.error("Failed to send group {} to tsd.", group, ex);
            }
        }
    }

    @Override
    public void run() {
        try {
            LOG.info("Start to run GroupInfoUpdater.");
            processBatch();
        } catch (Throwable t) {
            LOG.error("Faile to run GroupInfoUpdater", t);
        }
    }
}

@@ -756,16 +756,120 @@
 upBean;%0A
+import com.pinterest.arcee.bean.SpotAutoScalingBean;%0Aimport com.pinterest.arcee.dao.SpotAutoScalingDAO;%0A
 import c
@@ -1001,24 +1001,75 @@
 oupInfoDAO;%0A
+import com.pinterest.deployservice.bean.ASGStatus;%0A
 import com.p
@@ -1686,16 +1686,67 @@
 Manager;
+%0A    private SpotAutoScalingDAO spotAutoScalingDAO;
 %0A%0A    pu
@@ -1997,24 +1997,86 @@
 pManager();%0A
+        spotAutoScalingDAO = context.getSpotAutoScalingDAO();%0A
         tags
@@ -2415,24 +2415,25 @@
 pName);%0A
+%0A
         
 AutoScal
@@ -2424,16 +2424,25 @@
         
+List%3CSpot
 AutoScal
@@ -2448,19 +2448,19 @@
 ling
-GroupBean a
+Bean%3E spotA
 utoS
@@ -2469,41 +2469,34 @@
 ling
-Group
 Bean
+s
  = 
-a
+spotA
 utoScal
-eGroupManager
+ingDAO
 .get
@@ -2515,44 +2515,19 @@
 roup
-InfoByName(String.format(%22%25s-spot%22, 
+sByCluster(
 grou
@@ -2532,17 +2532,16 @@
 oupName)
-)
 ;%0A      
@@ -2605,40 +2605,340 @@
 if (
-autoScalingGroupBean != null) %7B%0A
+!spotAutoScalingBeans.isEmpty()) %7B%0A            String spotAutoScalingName = spotAutoScalingBeans.get(0).getAsg_name();%0A            AutoScalingGroupBean autoScalingGroupBean = autoScaleGroupManager.getAutoScalingGroupInfoByName(spotAutoScalingName);%0A            if (!autoScalingGroupBean.getStatus().equals(ASGStatus.UNKNOWN)) %7B%0A    
     
@@ -3013,24 +3013,28 @@
 s().size();%0A
+    
             
@@ -3080,32 +3080,36 @@
 nt;%0A            
+    
 metricSource.exp
@@ -3144,51 +3144,71 @@
 ME, 
-String.format(%22%25s-spot%22, group
+spotAutoScaling
 Name)
-)
 , tags,
+%0A                                   
  spo
@@ -3247,24 +3247,38 @@
 rrentTime);%0A
+            %7D%0A
         %7D%0A  
@@ -3733,108 +3733,8 @@
 p);%0A
-                // sleep 500 milliseconds for the next operation%0A                Thread.sleep(500);%0A
     

change group size info update frequency to make it query aws less