/*
 * 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 org.apache.brooklyn.entity.nosql.cassandra;

import java.math.BigDecimal;
import java.math.BigInteger;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.annotation.Nullable;

import org.apache.brooklyn.api.entity.Entity;
import org.apache.brooklyn.api.entity.EntitySpec;
import org.apache.brooklyn.api.location.Location;
import org.apache.brooklyn.api.policy.PolicySpec;
import org.apache.brooklyn.api.sensor.AttributeSensor;
import org.apache.brooklyn.api.sensor.SensorEvent;
import org.apache.brooklyn.api.sensor.SensorEventListener;
import org.apache.brooklyn.core.effector.EffectorBody;
import org.apache.brooklyn.core.entity.Attributes;
import org.apache.brooklyn.core.entity.Entities;
import org.apache.brooklyn.core.entity.EntityPredicates;
import org.apache.brooklyn.core.entity.lifecycle.Lifecycle;
import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic.ServiceNotUpLogic;
import org.apache.brooklyn.core.location.Machines;
import org.apache.brooklyn.enricher.stock.Enrichers;
import org.apache.brooklyn.entity.group.AbstractMembershipTrackingPolicy;
import org.apache.brooklyn.entity.group.DynamicClusterImpl;
import org.apache.brooklyn.entity.group.DynamicGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.brooklyn.util.collections.MutableList;
import org.apache.brooklyn.util.collections.MutableMap;
import org.apache.brooklyn.util.collections.MutableSet;
import org.apache.brooklyn.util.core.ResourceUtils;
import org.apache.brooklyn.util.core.config.ConfigBag;
import org.apache.brooklyn.util.text.Strings;
import org.apache.brooklyn.util.time.Time;

import com.google.common.base.Objects;
import com.google.common.base.Optional;
import com.google.common.base.Supplier;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.LinkedHashMultimap;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import com.google.common.net.HostAndPort;

/**
 * Implementation of {@link CassandraDatacenter}.
 * <p>
 * Several subtleties to note:
 * - a node may take some time after it is running and serving JMX to actually be contactable on its thrift port
 *   (so we wait for thrift port to be contactable)
 * - sometimes new nodes take a while to peer, and/or take a while to get a consistent schema
 *   (each up to 1m; often very close to the 1m) 
 */
public class CassandraDatacenterImpl extends DynamicClusterImpl implements CassandraDatacenter {

    /*
     * TODO Seed management is hard!
     *  - The ServiceRestarter is not doing customize(), so is not refreshing the seeds in cassandra.yaml.
     *    If we have two nodes that were seeds for each other and they both restart at the same time, we'll have a split brain.
     */
    
    private static final Logger log = LoggerFactory.getLogger(CassandraDatacenterImpl.class);

    // Mutex for synchronizing during re-size operations
    private final Object mutex = new Object[0];

    private final Supplier<Set<Entity>> defaultSeedSupplier = new Supplier<Set<Entity>>() {
        // Mutex for (re)calculating our seeds
        // TODO is this very dangerous?! Calling out to SeedTracker, which calls out to alien getAttribute()/getConfig(). But I think that's ok.
        // TODO might not need mutex? previous race was being caused by something else, other than concurrent calls!
        private final Object seedMutex = new Object();
        
        @Override
        public Set<Entity> get() {
            synchronized (seedMutex) {
                boolean hasPublishedSeeds = Boolean.TRUE.equals(getAttribute(HAS_PUBLISHED_SEEDS));
                int quorumSize = getSeedQuorumSize();
                Set<Entity> potentialSeeds = gatherPotentialSeeds();
                Set<Entity> potentialRunningSeeds = gatherPotentialRunningSeeds();
                boolean stillWaitingForQuorum = (!hasPublishedSeeds) && (potentialSeeds.size() < quorumSize);
                
                if (stillWaitingForQuorum) {
                    if (log.isDebugEnabled()) log.debug("Not refreshed seeds of cluster {}, because still waiting for quorum (need {}; have {} potentials)", new Object[] {CassandraDatacenterImpl.class, quorumSize, potentialSeeds.size()});
                    return ImmutableSet.of();
                } else if (hasPublishedSeeds) {
                    Set<Entity> currentSeeds = getAttribute(CURRENT_SEEDS);
                    if (getAttribute(SERVICE_STATE_ACTUAL) == Lifecycle.STARTING) {
                        if (Sets.intersection(currentSeeds, potentialSeeds).isEmpty()) {
                            log.warn("Cluster {} lost all its seeds while starting! Subsequent failure likely, but changing seeds during startup would risk split-brain: seeds={}", new Object[] {CassandraDatacenterImpl.this, currentSeeds});
                        }
                        return currentSeeds;
                    } else if (potentialRunningSeeds.isEmpty()) {
                        // TODO Could be race where nodes have only just returned from start() and are about to 
                        // transition to serviceUp; so don't just abandon all our seeds!
                        log.warn("Cluster {} has no running seeds (yet?); leaving seeds as-is; but risks split-brain if these seeds come back up!", new Object[] {CassandraDatacenterImpl.this});
                        return currentSeeds;
                    } else {
                        Set<Entity> result = trim(quorumSize, potentialRunningSeeds);
                        log.debug("Cluster {} updating seeds: chosen={}; potentialRunning={}", new Object[] {CassandraDatacenterImpl.this, result, potentialRunningSeeds});
                        return result;
                    }
                } else {
                    Set<Entity> result = trim(quorumSize, potentialSeeds);
                    if (log.isDebugEnabled()) log.debug("Cluster {} has reached seed quorum: seeds={}", new Object[] {CassandraDatacenterImpl.this, result});
                    return result;
                }
            }
        }
        private Set<Entity> trim(int num, Set<Entity> contenders) {
            // Prefer existing seeds wherever possible; otherwise accept any other contenders
            Set<Entity> currentSeeds = (getAttribute(CURRENT_SEEDS) != null) ? getAttribute(CURRENT_SEEDS) : ImmutableSet.<Entity>of();
            Set<Entity> result = Sets.newLinkedHashSet();
            result.addAll(Sets.intersection(currentSeeds, contenders));
            result.addAll(contenders);
            return ImmutableSet.copyOf(Iterables.limit(result, num));
        }
    };
    
    protected SeedTracker seedTracker = new SeedTracker();
    protected TokenGenerator tokenGenerator = null;

    public CassandraDatacenterImpl() {
    }

    @Override
    public void init() {
        super.init();

        /*
         * subscribe to hostname, and keep an accurate set of current seeds in a sensor;
         * then at nodes we set the initial seeds to be the current seeds when ready (non-empty)
         */
        subscriptions().subscribeToMembers(this, Attributes.HOSTNAME, new SensorEventListener<String>() {
            @Override
            public void onEvent(SensorEvent<String> event) {
                seedTracker.onHostnameChanged(event.getSource(), event.getValue());
            }
        });
        subscriptions().subscribe(this, DynamicGroup.MEMBER_REMOVED, new SensorEventListener<Entity>() {
            @Override public void onEvent(SensorEvent<Entity> event) {
                seedTracker.onMemberRemoved(event.getValue());
            }
        });
        subscriptions().subscribeToMembers(this, Attributes.SERVICE_UP, new SensorEventListener<Boolean>() {
            @Override
            public void onEvent(SensorEvent<Boolean> event) {
                seedTracker.onServiceUpChanged(event.getSource(), event.getValue());
            }
        });
        subscriptions().subscribeToMembers(this, Attributes.SERVICE_STATE_ACTUAL, new SensorEventListener<Lifecycle>() {
            @Override
            public void onEvent(SensorEvent<Lifecycle> event) {
                // trigger a recomputation also when lifecycle state changes, 
                // because it might not have ruled a seed as inviable when service up went true 
                // because service state was not yet running
                seedTracker.onServiceUpChanged(event.getSource(), Lifecycle.RUNNING==event.getValue());
            }
        });
        
        // Track the datacenters for this cluster
        subscriptions().subscribeToMembers(this, CassandraNode.DATACENTER_NAME, new SensorEventListener<String>() {
            @Override
            public void onEvent(SensorEvent<String> event) {
                Entity member = event.getSource();
                String dcName = event.getValue();
                if (dcName != null) {
                    Multimap<String, Entity> datacenterUsage = getAttribute(DATACENTER_USAGE);
                    Multimap<String, Entity> mutableDatacenterUsage = (datacenterUsage == null) ? LinkedHashMultimap.<String, Entity>create() : LinkedHashMultimap.create(datacenterUsage);
                    Optional<String> oldDcName = getKeyOfVal(mutableDatacenterUsage, member);
                    if (!(oldDcName.isPresent() && dcName.equals(oldDcName.get()))) {
                        mutableDatacenterUsage.values().remove(member);
                        mutableDatacenterUsage.put(dcName, member);
                        sensors().set(DATACENTER_USAGE, mutableDatacenterUsage);
                        sensors().set(DATACENTERS, Sets.newLinkedHashSet(mutableDatacenterUsage.keySet()));
                    }
                }
            }
            private <K,V> Optional<K> getKeyOfVal(Multimap<K,V> map, V val) {
                for (Map.Entry<K,V> entry : map.entries()) {
                    if (Objects.equal(val, entry.getValue())) {
                        return Optional.of(entry.getKey());
                    }
                }
                return Optional.absent();
            }
        });
        subscriptions().subscribe(this, DynamicGroup.MEMBER_REMOVED, new SensorEventListener<Entity>() {
            @Override public void onEvent(SensorEvent<Entity> event) {
                Entity entity = event.getSource();
                Multimap<String, Entity> datacenterUsage = getAttribute(DATACENTER_USAGE);
                if (datacenterUsage != null && datacenterUsage.containsValue(entity)) {
                    Multimap<String, Entity> mutableDatacenterUsage = LinkedHashMultimap.create(datacenterUsage);
                    mutableDatacenterUsage.values().remove(entity);
                    sensors().set(DATACENTER_USAGE, mutableDatacenterUsage);
                    sensors().set(DATACENTERS, Sets.newLinkedHashSet(mutableDatacenterUsage.keySet()));
                }
            }
        });
        
        getMutableEntityType().addEffector(EXECUTE_SCRIPT, new EffectorBody<String>() {
            @Override
            public String call(ConfigBag parameters) {
                return executeScript((String)parameters.getStringKey("commands"));
            }
        });
    }
    
    protected Supplier<Set<Entity>> getSeedSupplier() {
        Supplier<Set<Entity>> seedSupplier = getConfig(SEED_SUPPLIER);
        return (seedSupplier == null) ? defaultSeedSupplier : seedSupplier;
    }
    
    protected boolean useVnodes() {
        return Boolean.TRUE.equals(getConfig(USE_VNODES));
    }
    
    protected synchronized TokenGenerator getTokenGenerator() {
        if (tokenGenerator!=null) 
            return tokenGenerator;
        
        try {
            tokenGenerator = getConfig(TOKEN_GENERATOR_CLASS).newInstance();
            
            BigInteger shift = getConfig(TOKEN_SHIFT);
            if (shift==null) 
                shift = BigDecimal.valueOf(Math.random()).multiply(
                    new BigDecimal(tokenGenerator.range())).toBigInteger();
            tokenGenerator.setOrigin(shift);
            
            return tokenGenerator;
        } catch (Exception e) {
            throw Throwables.propagate(e);
        }        
    }
    
    protected int getSeedQuorumSize() {
        Integer quorumSize = getConfig(INITIAL_QUORUM_SIZE);
        if (quorumSize!=null && quorumSize>0)
            return quorumSize;
        // default 2 is recommended, unless initial size is smaller
        return Math.min(Math.max(getConfig(INITIAL_SIZE), 1), DEFAULT_SEED_QUORUM);
    }

    @Override
    public Set<Entity> gatherPotentialSeeds() {
        return seedTracker.gatherPotentialSeeds();
    }

    @Override
    public Set<Entity> gatherPotentialRunningSeeds() {
        return seedTracker.gatherPotentialRunningSeeds();
    }

    /**
     * Sets the default {@link #MEMBER_SPEC} to describe the Cassandra nodes.
     */
    @Override
    protected EntitySpec<?> getMemberSpec() {
        return getConfig(MEMBER_SPEC, EntitySpec.create(CassandraNode.class));
    }

    @Override
    public String getClusterName() {
        return getAttribute(CLUSTER_NAME);
    }

    @Override
    public Collection<Entity> grow(int delta) {
        if (useVnodes()) {
            // nothing to do for token generator
        } else {
            if (getCurrentSize() == 0) {
                getTokenGenerator().growingCluster(delta);
            }
        }
        return super.grow(delta);
    }
    
    @Override
    protected Entity createNode(@Nullable Location loc, Map<?,?> flags) {
        Map<Object, Object> allflags = MutableMap.copyOf(flags);

        if (flags.containsKey("token") || flags.containsKey("cassandra.token")) {
            // TODO Delete in future version; was deprecated in 0.7.0; deleted config key in 0.9.0
            log.warn("Cassandra token no longer supported - use 'tokens' in "+CassandraDatacenterImpl.this);
        }
        if (flags.containsKey(CassandraNode.TOKENS) || flags.containsKey("tokens") || flags.containsKey("cassandra.tokens")) {
            // leave token config as-is
        } else if (!useVnodes()) {
            BigInteger token = getTokenGenerator().newToken();
            if (token != null) {
                allflags.put(CassandraNode.TOKENS, ImmutableSet.of(token));
            }
        }

        if ((flags.containsKey(CassandraNode.NUM_TOKENS_PER_NODE) || flags.containsKey("numTokensPerNode"))) {
            // leave num_tokens as-is
        } else if (useVnodes()) {
            Integer numTokensPerNode = getConfig(NUM_TOKENS_PER_NODE);
            allflags.put(CassandraNode.NUM_TOKENS_PER_NODE, numTokensPerNode);
        } else {
            allflags.put(CassandraNode.NUM_TOKENS_PER_NODE, 1);
        }
        
        return super.createNode(loc, allflags);
    }

    @Override
    protected Entity replaceMember(Entity member, Location memberLoc, Map<?, ?> extraFlags) {
        Set<BigInteger> oldTokens = ((CassandraNode) member).getTokens();
        Set<BigInteger> newTokens = (oldTokens != null && oldTokens.size() > 0) ? getTokenGenerator().getTokensForReplacementNode(oldTokens) : null;
        MutableMap<Object, Object> allFlags = MutableMap.<Object, Object>copyOf(extraFlags).add(CassandraNode.TOKENS, newTokens);
        return super.replaceMember(member, memberLoc, allFlags);
    }

    @Override
    public void start(Collection<? extends Location> locations) {
        Machines.warnIfLocalhost(locations, "CassandraCluster does not support multiple nodes on localhost, " +
                "due to assumptions Cassandra makes about the use of the same port numbers used across the cluster.");

        // force this to be set - even if it is using the default
        sensors().set(CLUSTER_NAME, getConfig(CLUSTER_NAME));
        
        super.start(locations);

        connectSensors();

        // TODO wait until all nodes which we think are up are consistent 
        // i.e. all known nodes use the same schema, as reported by
        // SshEffectorTasks.ssh("echo \"describe cluster;\" | /bin/cassandra-cli");
        // once we've done that we can revert to using 2 seed nodes.
        // see CassandraCluster.DEFAULT_SEED_QUORUM
        // (also ensure the cluster is ready if we are about to run a creation script)
        Time.sleep(getConfig(DELAY_BEFORE_ADVERTISING_CLUSTER));

        String scriptUrl = getConfig(CassandraNode.CREATION_SCRIPT_URL);
        if (Strings.isNonEmpty(scriptUrl)) {
            executeScript(new ResourceUtils(this).getResourceAsString(scriptUrl));
        }

        update();
    }

    protected void connectSensors() {
        connectEnrichers();
        
        policies().add(PolicySpec.create(MemberTrackingPolicy.class)
                .displayName("Cassandra Cluster Tracker")
                .configure("sensorsToTrack", ImmutableSet.of(Attributes.SERVICE_UP, Attributes.HOSTNAME, CassandraNode.THRIFT_PORT))
                .configure("group", this));
    }

    public static class MemberTrackingPolicy extends AbstractMembershipTrackingPolicy {
        @Override
        protected void onEntityChange(Entity member) {
            if (log.isDebugEnabled()) log.debug("Node {} updated in Cluster {}", member, this);
            ((CassandraDatacenterImpl)entity).update();
        }
        @Override
        protected void onEntityAdded(Entity member) {
            if (log.isDebugEnabled()) log.debug("Node {} added to Cluster {}", member, this);
            ((CassandraDatacenterImpl)entity).update();
        }
        @Override
        protected void onEntityRemoved(Entity member) {
            if (log.isDebugEnabled()) log.debug("Node {} removed from Cluster {}", member, this);
            ((CassandraDatacenterImpl)entity).update();
        }
    };

    @SuppressWarnings("unchecked")
    protected void connectEnrichers() {
        List<? extends List<? extends AttributeSensor<? extends Number>>> summingEnricherSetup = ImmutableList.of(
                ImmutableList.of(CassandraNode.READ_ACTIVE, READ_ACTIVE),
                ImmutableList.of(CassandraNode.READ_PENDING, READ_PENDING),
                ImmutableList.of(CassandraNode.WRITE_ACTIVE, WRITE_ACTIVE),
                ImmutableList.of(CassandraNode.WRITE_PENDING, WRITE_PENDING)
        );
        
        List<? extends List<? extends AttributeSensor<? extends Number>>> averagingEnricherSetup = ImmutableList.of(
                ImmutableList.of(CassandraNode.READS_PER_SECOND_LAST, READS_PER_SECOND_LAST_PER_NODE),
                ImmutableList.of(CassandraNode.WRITES_PER_SECOND_LAST, WRITES_PER_SECOND_LAST_PER_NODE),
                ImmutableList.of(CassandraNode.WRITES_PER_SECOND_IN_WINDOW, WRITES_PER_SECOND_IN_WINDOW_PER_NODE),
                ImmutableList.of(CassandraNode.READS_PER_SECOND_IN_WINDOW, READS_PER_SECOND_IN_WINDOW_PER_NODE),
                ImmutableList.of(CassandraNode.THRIFT_PORT_LATENCY, THRIFT_PORT_LATENCY_PER_NODE),
                ImmutableList.of(CassandraNode.THRIFT_PORT_LATENCY_IN_WINDOW, THRIFT_PORT_LATENCY_IN_WINDOW_PER_NODE),
                ImmutableList.of(CassandraNode.PROCESS_CPU_TIME_FRACTION_LAST, PROCESS_CPU_TIME_FRACTION_LAST_PER_NODE),
                ImmutableList.of(CassandraNode.PROCESS_CPU_TIME_FRACTION_IN_WINDOW, PROCESS_CPU_TIME_FRACTION_IN_WINDOW_PER_NODE)
        );
        
        for (List<? extends AttributeSensor<? extends Number>> es : summingEnricherSetup) {
            AttributeSensor<? extends Number> t = es.get(0);
            AttributeSensor<? extends Number> total = es.get(1);
            enrichers().add(Enrichers.builder()
                    .aggregating(t)
                    .publishing(total)
                    .fromMembers()
                    .computingSum()
                    .defaultValueForUnreportedSensors(null)
                    .valueToReportIfNoSensors(null)
                    .build());
        }
        
        for (List<? extends AttributeSensor<? extends Number>> es : averagingEnricherSetup) {
            AttributeSensor<Number> t = (AttributeSensor<Number>) es.get(0);
            AttributeSensor<Double> average = (AttributeSensor<Double>) es.get(1);
            enrichers().add(Enrichers.builder()
                    .aggregating(t)
                    .publishing(average)
                    .fromMembers()
                    .computingAverage()
                    .defaultValueForUnreportedSensors(null)
                    .valueToReportIfNoSensors(null)
                    .build());

        }
    }

    @Override
    public void stop() {
        disconnectSensors();
        
        super.stop();
    }
    
    protected void disconnectSensors() {
    }

    @Override
    public void update() {
        synchronized (mutex) {
            // Update our seeds, as necessary
            seedTracker.refreshSeeds();
            
            // Choose the first available cluster member to set host and port (and compute one-up)
            Optional<Entity> upNode = Iterables.tryFind(getMembers(), EntityPredicates.attributeEqualTo(SERVICE_UP, Boolean.TRUE));

            if (upNode.isPresent()) {
                sensors().set(HOSTNAME, upNode.get().getAttribute(Attributes.HOSTNAME));
                sensors().set(THRIFT_PORT, upNode.get().getAttribute(CassandraNode.THRIFT_PORT));

                List<String> currentNodes = getAttribute(CASSANDRA_CLUSTER_NODES);
                Set<String> oldNodes = (currentNodes != null) ? ImmutableSet.copyOf(currentNodes) : ImmutableSet.<String>of();
                Set<String> newNodes = MutableSet.<String>of();
                for (Entity member : getMembers()) {
                    if (member instanceof CassandraNode && Boolean.TRUE.equals(member.getAttribute(SERVICE_UP))) {
                        String hostname = member.getAttribute(Attributes.HOSTNAME);
                        Integer thriftPort = member.getAttribute(CassandraNode.THRIFT_PORT);
                        if (hostname != null && thriftPort != null) {
                            newNodes.add(HostAndPort.fromParts(hostname, thriftPort).toString());
                        }
                    }
                }
                if (Sets.symmetricDifference(oldNodes, newNodes).size() > 0) {
                    sensors().set(CASSANDRA_CLUSTER_NODES, MutableList.copyOf(newNodes));
                }
            } else {
                sensors().set(HOSTNAME, null);
                sensors().set(THRIFT_PORT, null);
                sensors().set(CASSANDRA_CLUSTER_NODES, Collections.<String>emptyList());
            }

            ServiceNotUpLogic.updateNotUpIndicatorRequiringNonEmptyList(this, CASSANDRA_CLUSTER_NODES);
        }
    }
    
    /**
     * For tracking our seeds. This gets fiddly! High-level logic is:
     * <ul>
     *   <li>If we have never reached quorum (i.e. have never published seeds), then continue to wait for quorum;
     *       because entity-startup may be blocking for this. This is handled by the seedSupplier.
     *   <li>If we previously reached quorum (i.e. have previousy published seeds), then always update;
     *       we never want stale/dead entities listed in our seeds.
     *   <li>If an existing seed looks unhealthy, then replace it.
     *   <li>If a new potential seed becomes available (and we're in need of more), then add it.
     * <ul>
     * 
     * Also note that {@link CassandraFabric} can take over, because it know about multiple sub-clusters!
     * It will provide a different {@link CassandraDatacenter#SEED_SUPPLIER}. Each time we think that our seeds
     * need to change, we call that. The fabric will call into {@link CassandraDatacenterImpl#gatherPotentialSeeds()}
     * to find out what's available.
     * 
     * @author aled
     */
    protected class SeedTracker {
        private final Map<Entity, Boolean> memberUpness = Maps.newLinkedHashMap();
        
        public void onMemberRemoved(Entity member) {
            Set<Entity> seeds = getSeeds();
            boolean maybeRemove = seeds.contains(member);
            memberUpness.remove(member);
            
            if (maybeRemove) {
                refreshSeeds();
            } else {
                if (log.isTraceEnabled()) log.trace("Seeds considered stable for cluster {} (node {} removed)", new Object[] {CassandraDatacenterImpl.this, member});
                return;
            }
        }
        public void onHostnameChanged(Entity member, String hostname) {
            Set<Entity> seeds = getSeeds();
            int quorum = getSeedQuorumSize();
            boolean isViable = isViableSeed(member);
            boolean maybeAdd = isViable && seeds.size() < quorum;
            boolean maybeRemove = seeds.contains(member) && !isViable;
            
            if (maybeAdd || maybeRemove) {
                refreshSeeds();
            } else {
                if (log.isTraceEnabled()) log.trace("Seeds considered stable for cluster {} (node {} changed hostname {})", new Object[] {CassandraDatacenterImpl.this, member, hostname});
                return;
            }
        }
        public void onServiceUpChanged(Entity member, Boolean serviceUp) {
            Boolean oldVal = memberUpness.put(member, serviceUp);
            if (Objects.equal(oldVal, serviceUp)) {
                if (log.isTraceEnabled()) log.trace("Ignoring duplicate service-up in "+CassandraDatacenterImpl.this+" for "+member+", "+serviceUp);
            }
            Set<Entity> seeds = getSeeds();
            int quorum = getSeedQuorumSize();
            boolean isViable = isViableSeed(member);
            boolean maybeAdd = isViable && seeds.size() < quorum;
            boolean maybeRemove = seeds.contains(member) && !isViable;
            
            if (log.isDebugEnabled())
                log.debug("Considering refresh of seeds for "+CassandraDatacenterImpl.this+" because "+member+" is now "+serviceUp+" ("+isViable+" / "+maybeAdd+" / "+maybeRemove+")");
            if (maybeAdd || maybeRemove) {
                refreshSeeds();
            } else {
                if (log.isTraceEnabled()) log.trace("Seeds considered stable for cluster {} (node {} changed serviceUp {})", new Object[] {CassandraDatacenterImpl.this, member, serviceUp});
                return;
            }
        }
        protected Set<Entity> getSeeds() {
            Set<Entity> result = getAttribute(CURRENT_SEEDS);
            return (result == null) ? ImmutableSet.<Entity>of() : result;
        }
        public void refreshSeeds() {
            Set<Entity> oldseeds = getAttribute(CURRENT_SEEDS);
            Set<Entity> newseeds = getSeedSupplier().get();
            if (Objects.equal(oldseeds, newseeds)) {
                if (log.isTraceEnabled()) log.debug("Seed refresh no-op for cluster {}: still={}", new Object[] {CassandraDatacenterImpl.this, oldseeds});
            } else {
                if (log.isDebugEnabled()) log.debug("Refreshing seeds of cluster {}: now={}; old={}", new Object[] {this, newseeds, oldseeds});
                sensors().set(CURRENT_SEEDS, newseeds);
                if (newseeds != null && newseeds.size() > 0) {
                    sensors().set(HAS_PUBLISHED_SEEDS, true);
                }
            }
        }
        public Set<Entity> gatherPotentialSeeds() {
            Set<Entity> result = Sets.newLinkedHashSet();
            for (Entity member : getMembers()) {
                if (isViableSeed(member)) {
                    result.add(member);
                }
            }
            if (log.isTraceEnabled()) log.trace("Viable seeds in Cluster {}: {}", new Object[] {result});
            return result;
        }
        public Set<Entity> gatherPotentialRunningSeeds() {
            Set<Entity> result = Sets.newLinkedHashSet();
            for (Entity member : getMembers()) {
                if (isRunningSeed(member)) {
                    result.add(member);
                }
            }
            if (log.isTraceEnabled()) log.trace("Viable running seeds in Cluster {}: {}", new Object[] {result});
            return result;
        }
        public boolean isViableSeed(Entity member) {
            // TODO would be good to reuse the better logic in ServiceFailureDetector
            // (e.g. if that didn't just emit a notification but set a sensor as well?)
            boolean managed = Entities.isManaged(member);
            String hostname = member.getAttribute(Attributes.HOSTNAME);
            boolean serviceUp = Boolean.TRUE.equals(member.getAttribute(Attributes.SERVICE_UP));
            Lifecycle serviceState = member.getAttribute(Attributes.SERVICE_STATE_ACTUAL);
            boolean hasFailed = !managed || (serviceState == Lifecycle.ON_FIRE) || (serviceState == Lifecycle.RUNNING && !serviceUp) || (serviceState == Lifecycle.STOPPED);
            boolean result = (hostname != null && !hasFailed);
            if (log.isTraceEnabled()) log.trace("Node {} in Cluster {}: viableSeed={}; hostname={}; serviceUp={}; serviceState={}; hasFailed={}", new Object[] {member, this, result, hostname, serviceUp, serviceState, hasFailed});
            return result;
        }
        public boolean isRunningSeed(Entity member) {
            boolean viableSeed = isViableSeed(member);
            boolean serviceUp = Boolean.TRUE.equals(member.getAttribute(Attributes.SERVICE_UP));
            Lifecycle serviceState = member.getAttribute(Attributes.SERVICE_STATE_ACTUAL);
            boolean result = viableSeed && serviceUp && serviceState == Lifecycle.RUNNING;
            if (log.isTraceEnabled()) log.trace("Node {} in Cluster {}: runningSeed={}; viableSeed={}; serviceUp={}; serviceState={}", new Object[] {member, this, result, viableSeed, serviceUp, serviceState});
            return result;
        }
    }
    
    @Override
    public String executeScript(String commands) {
        Entity someChild = Iterables.getFirst(getMembers(), null);
        if (someChild==null)
            throw new IllegalStateException("No Cassandra nodes available");
        // FIXME cross-etntity method-style calls such as below do not set up a queueing context (DynamicSequentialTask) 
//        return ((CassandraNode)someChild).executeScript(commands);
        return Entities.invokeEffector(this, someChild, CassandraNode.EXECUTE_SCRIPT, MutableMap.of("commands", commands)).getUnchecked();
    }
    
}