<commit_before>/**
 * Copyright 2013 Netflix, 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.netflix.dynomitemanager.identity;

import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;

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

import com.google.common.base.Predicate;
import com.google.common.base.Supplier;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimaps;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.netflix.dynomitemanager.sidecore.IConfiguration;
import com.netflix.dynomitemanager.sidecore.utils.ITokenManager;
import com.netflix.dynomitemanager.sidecore.utils.RetryableCallable;
import com.netflix.dynomitemanager.sidecore.utils.Sleeper;
import com.netflix.dynomitemanager.identity.AppsInstance;
import com.netflix.dynomitemanager.identity.IAppsInstanceFactory;
import com.netflix.dynomitemanager.identity.IMembership;
import com.netflix.dynomitemanager.identity.InstanceEnvIdentity;
import com.netflix.dynomitemanager.identity.InstanceIdentity;

/**
 * This class provides the central place to create and consume the identity of
 * the instance - token, seeds etc.
 * 
 */

@Singleton
public class InstanceIdentity
{
    private static final Logger logger = LoggerFactory.getLogger(InstanceIdentity.class);
    private static final String DUMMY_INSTANCE_ID = "new_slot";

    private final ListMultimap<String, AppsInstance> locMap = Multimaps.newListMultimap(new HashMap<String, Collection<AppsInstance>>(), new Supplier<List<AppsInstance>>()
    {
        public List<AppsInstance> get()
        {
            return Lists.newArrayList();
        }
    });
    private final IAppsInstanceFactory factory;
    private final IMembership membership;
    private final IConfiguration config;
    private final Sleeper sleeper;
    private final ITokenManager tokenManager;
	private final InstanceEnvIdentity insEnvIdentity;    


    private final Predicate<AppsInstance> differentHostPredicate = new Predicate<AppsInstance>() {
    		@Override
    		public boolean apply(AppsInstance instance) {
    			return (!instance.getInstanceId().equalsIgnoreCase(DUMMY_INSTANCE_ID) && !instance.getHostName().equals(myInstance.getHostName()));
    		}
    };
   
    private AppsInstance myInstance;
    private boolean isReplace = false;
    private boolean isTokenPregenerated = false;
    private String replacedIp = "";

    @Inject
    public InstanceIdentity(IAppsInstanceFactory factory, IMembership membership, IConfiguration config,
                            Sleeper sleeper, ITokenManager tokenManager, InstanceEnvIdentity insEnvIdentity) throws Exception
    {
        this.factory = factory;
        this.membership = membership;
        this.config = config;
        this.sleeper = sleeper;
        this.tokenManager = tokenManager;
        this.insEnvIdentity = insEnvIdentity;
        init();
    }

    public AppsInstance getInstance()
    {
        return myInstance;
    }

    public void init() throws Exception
    {
        // try to grab the token which was already assigned
        myInstance = new RetryableCallable<AppsInstance>()
        {
            @Override
            public AppsInstance retriableCall() throws Exception
            {
                // Check if this node is decommissioned
                for (AppsInstance ins : factory.getAllIds(config.getAppName() + "-dead"))
                {
                    logger.debug(String.format("[Dead] Iterating though the hosts: %s", ins.getInstanceId()));
                    if (ins.getInstanceId().equals(config.getInstanceName()))
                    {
                        ins.setOutOfService(true);
                        return ins;
                    }
                }
                for (AppsInstance ins : factory.getAllIds(config.getAppName()))
                {
                    logger.debug(String.format("[Alive] Iterating though the hosts: %s My id = [%s]", ins.getInstanceId(),ins.getId()));
                    if (ins.getInstanceId().equals(config.getInstanceName()))
                        return ins;
                }
                return null;
            }
        }.call();
        // Grab a dead token
        if (null == myInstance)
            myInstance = new GetDeadToken().call();
        
        // Grab a pre-generated token if there is such one
        if (null == myInstance)
           myInstance = new GetPregeneratedToken().call();
        
        // Grab a new token
        if (null == myInstance)
        {
			GetNewToken newToken = new GetNewToken();
			newToken.set(100, 100);
			myInstance = newToken.call();
		}
        logger.info("My token: " + myInstance.getToken());
        
    }

    private void populateRacMap()
    {
        locMap.clear();
        for (AppsInstance ins : factory.getAllIds(config.getAppName()))
        {
        		locMap.put(ins.getZone(), ins);
        }
    }

    public class GetDeadToken extends RetryableCallable<AppsInstance>
    {
        @Override
        public AppsInstance retriableCall() throws Exception
        {
            final List<AppsInstance> allIds = factory.getAllIds(config.getAppName());
            List<String> asgInstances = membership.getRacMembership();
            if (config.isDualAccount()){
            	logger.info("Dual Account cluster");
            	List<String> crossAccountAsgInstances = membership.getCrossAccountRacMembership();           

            	if (insEnvIdentity.isClassic()) {
            		logger.info("EC2 classic instances (local ASG): " + Arrays.toString(asgInstances.toArray()));
            		logger.info("VPC Account (cross-account ASG): " + Arrays.toString(crossAccountAsgInstances.toArray()));
            	}
            	else {
            		logger.info("VPC Account (local ASG): " + Arrays.toString(asgInstances.toArray()));
            		logger.info("EC2 classic instances (cross-account ASG): " + Arrays.toString(crossAccountAsgInstances.toArray()));
            	}
          
            	// Remove duplicates (probably there are not)
            	asgInstances.removeAll(crossAccountAsgInstances);
            
            	// Merge the two lists
            	asgInstances.addAll(crossAccountAsgInstances);
            	logger.info("Combined Instances in the AZ: " + asgInstances);
            }
            
            // Sleep random interval - upto 15 sec
            sleeper.sleep(new Random().nextInt(5000) + 10000);
            for (AppsInstance dead : allIds)
            {
                // test same dc and is it is alive.
                if (!dead.getRack().equals(config.getRack()) || asgInstances.contains(dead.getInstanceId()))
                    continue;
                logger.info("Found dead instances: " + dead.getInstanceId());
                //AppsInstance markAsDead = factory.create(dead.getApp() + "-dead", dead.getId(), dead.getInstanceId(), dead.getHostName(), dead.getHostIP(), dead.getZone(), 
                //                                         dead.getVolumes(), dead.getToken(), dead.getRack());
                // remove it as we marked it down...
                //factory.delete(dead);
                isReplace = true;
                replacedIp = dead.getHostIP();
                String payLoad = dead.getToken();
                logger.info("Trying to grab slot {} with availability zone {}", dead.getId(), dead.getZone());
                return factory.create(config.getAppName(), dead.getId(), config.getInstanceName(), config.getHostname(), config.getHostIP(), 
                                      config.getZone(), dead.getVolumes(), payLoad, config.getRack());
            }
            return null;
        }
<commit_msg>removing the pollutant in the GetDeadToken class to a seperate class<commit_after>/**
 * Copyright 2013 Netflix, 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.netflix.dynomitemanager.identity;

import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;

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

import com.google.common.base.Predicate;
import com.google.common.base.Supplier;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimaps;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.netflix.dynomitemanager.sidecore.IConfiguration;
import com.netflix.dynomitemanager.sidecore.utils.ITokenManager;
import com.netflix.dynomitemanager.sidecore.utils.RetryableCallable;
import com.netflix.dynomitemanager.sidecore.utils.Sleeper;
import com.netflix.dynomitemanager.identity.AppsInstance;
import com.netflix.dynomitemanager.identity.IAppsInstanceFactory;
import com.netflix.dynomitemanager.identity.IMembership;
import com.netflix.dynomitemanager.identity.InstanceEnvIdentity;
import com.netflix.dynomitemanager.identity.InstanceIdentity;

/**
 * This class provides the central place to create and consume the identity of
 * the instance - token, seeds etc.
 * 
 */

@Singleton
public class InstanceIdentity
{
    private static final Logger logger = LoggerFactory.getLogger(InstanceIdentity.class);
    private static final String DUMMY_INSTANCE_ID = "new_slot";

    private final ListMultimap<String, AppsInstance> locMap = Multimaps.newListMultimap(new HashMap<String, Collection<AppsInstance>>(), new Supplier<List<AppsInstance>>()
    {
        public List<AppsInstance> get()
        {
            return Lists.newArrayList();
        }
    });
    private final IAppsInstanceFactory factory;
    private final IMembership membership;
    private final IConfiguration config;
    private final Sleeper sleeper;
    private final ITokenManager tokenManager;
	private final InstanceEnvIdentity insEnvIdentity;    


    private final Predicate<AppsInstance> differentHostPredicate = new Predicate<AppsInstance>() {
    		@Override
    		public boolean apply(AppsInstance instance) {
    			return (!instance.getInstanceId().equalsIgnoreCase(DUMMY_INSTANCE_ID) && !instance.getHostName().equals(myInstance.getHostName()));
    		}
    };
   
    private AppsInstance myInstance;
    private boolean isReplace = false;
    private boolean isTokenPregenerated = false;
    private String replacedIp = "";

    @Inject
    public InstanceIdentity(IAppsInstanceFactory factory, IMembership membership, IConfiguration config,
                            Sleeper sleeper, ITokenManager tokenManager, InstanceEnvIdentity insEnvIdentity) throws Exception
    {
        this.factory = factory;
        this.membership = membership;
        this.config = config;
        this.sleeper = sleeper;
        this.tokenManager = tokenManager;
        this.insEnvIdentity = insEnvIdentity;
        init();
    }

    public AppsInstance getInstance()
    {
        return myInstance;
    }

    public void init() throws Exception
    {
        // try to grab the token which was already assigned
        myInstance = new RetryableCallable<AppsInstance>()
        {
            @Override
            public AppsInstance retriableCall() throws Exception
            {
                // Check if this node is decommissioned
                for (AppsInstance ins : factory.getAllIds(config.getAppName() + "-dead"))
                {
                    logger.debug(String.format("[Dead] Iterating though the hosts: %s", ins.getInstanceId()));
                    if (ins.getInstanceId().equals(config.getInstanceName()))
                    {
                        ins.setOutOfService(true);
                        return ins;
                    }
                }
                for (AppsInstance ins : factory.getAllIds(config.getAppName()))
                {
                    logger.debug(String.format("[Alive] Iterating though the hosts: %s My id = [%s]", ins.getInstanceId(),ins.getId()));
                    if (ins.getInstanceId().equals(config.getInstanceName()))
                        return ins;
                }
                return null;
            }
        }.call();
        // Grab a dead token
        if (null == myInstance)
            myInstance = new GetDeadToken().call();
        
        // Grab a pre-generated token if there is such one
        if (null == myInstance)
           myInstance = new GetPregeneratedToken().call();
        
        // Grab a new token
        if (null == myInstance)
        {
			GetNewToken newToken = new GetNewToken();
			newToken.set(100, 100);
			myInstance = newToken.call();
		}
        logger.info("My token: " + myInstance.getToken());
        
    }

    private void populateRacMap()
    {
        locMap.clear();
        for (AppsInstance ins : factory.getAllIds(config.getAppName()))
        {
        		locMap.put(ins.getZone(), ins);
        }
    }

    private List<String> getDualAccountRacMembership(List<String> asgInstances){
    	logger.info("Dual Account cluster");

        List<String> crossAccountAsgInstances = membership.getCrossAccountRacMembership();           

        if (insEnvIdentity.isClassic()) {
           logger.info("EC2 classic instances (local ASG): " + Arrays.toString(asgInstances.toArray()));
           logger.info("VPC Account (cross-account ASG): " + Arrays.toString(crossAccountAsgInstances.toArray()));
        }
        else {
           logger.info("VPC Account (local ASG): " + Arrays.toString(asgInstances.toArray()));
           logger.info("EC2 classic instances (cross-account ASG): " + Arrays.toString(crossAccountAsgInstances.toArray()));
        }
  
        // Remove duplicates (probably there are not)
        asgInstances.removeAll(crossAccountAsgInstances);
    
        // Merge the two lists
        asgInstances.addAll(crossAccountAsgInstances);
        logger.info("Combined Instances in the AZ: " + asgInstances);
        
    	return asgInstances;
    }

    public class GetDeadToken extends RetryableCallable<AppsInstance>
    {
        @Override
        public AppsInstance retriableCall() throws Exception
        {
            final List<AppsInstance> allIds = factory.getAllIds(config.getAppName());
            List<String> asgInstances = membership.getRacMembership();       
            if (config.isDualAccount()){
            	asgInstances = getDualAccountRacMembership(asgInstances);
            }
            else {
            	logger.info("Single Account cluster");
            }
            
            // Sleep random interval - upto 15 sec
            sleeper.sleep(new Random().nextInt(5000) + 10000);
            for (AppsInstance dead : allIds)
            {
                // test same dc and is it is alive.
                if (!dead.getRack().equals(config.getRack()) || asgInstances.contains(dead.getInstanceId()))
                    continue;
                logger.info("Found dead instances: " + dead.getInstanceId());
                //AppsInstance markAsDead = factory.create(dead.getApp() + "-dead", dead.getId(), dead.getInstanceId(), dead.getHostName(), dead.getHostIP(), dead.getZone(), 
                //                                         dead.getVolumes(), dead.getToken(), dead.getRack());
                // remove it as we marked it down...
                //factory.delete(dead);
                isReplace = true;
                replacedIp = dead.getHostIP();
                String payLoad = dead.getToken();
                logger.info("Trying to grab slot {} with availability zone {}", dead.getId(), dead.getZone());
                return factory.create(config.getAppName(), dead.getId(), config.getInstanceName(), config.getHostname(), config.getHostIP(), 
                                      config.getZone(), dead.getVolumes(), payLoad, config.getRack());
            }
            return null;
        }

{'fraction_non_alphanumeric': 0.09884264671498715, 'fraction_numerical': 0.0033902267944821134, 'mean_word_length': 2.4252252252252253, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '144425', 'n_tokens_mistral': 4640, 'n_tokens_neox': 4499, 'n_words': 1094}