gerrit.opnfv Code Review - onosfw.git/blob

   1 package org.onosproject.store.cluster.impl;
   2
   3 import static com.google.common.base.Preconditions.checkNotNull;
   4 import static java.net.NetworkInterface.getNetworkInterfaces;
   5 import static org.slf4j.LoggerFactory.getLogger;
   6
   7 import java.io.File;
   8 import java.io.IOException;
   9 import java.net.InetAddress;
  10 import java.net.NetworkInterface;
  11 import java.util.Arrays;
  12 import java.util.Collection;
  13 import java.util.Collections;
  14 import java.util.concurrent.atomic.AtomicReference;
  15 import java.util.function.Function;
  16
  17 import org.apache.felix.scr.annotations.Activate;
  18 import org.apache.felix.scr.annotations.Component;
  19 import org.apache.felix.scr.annotations.Deactivate;
  20 import org.apache.felix.scr.annotations.Service;
  21 import org.onlab.packet.IpAddress;
  22 import org.onosproject.cluster.ClusterMetadata;
  23 import org.onosproject.cluster.ClusterMetadataEvent;
  24 import org.onosproject.cluster.ClusterMetadataStore;
  25 import org.onosproject.cluster.ClusterMetadataStoreDelegate;
  26 import org.onosproject.cluster.ControllerNode;
  27 import org.onosproject.cluster.DefaultControllerNode;
  28 import org.onosproject.cluster.NodeId;
  29 import org.onosproject.cluster.Partition;
  30 import org.onosproject.store.AbstractStore;
  31 import org.onosproject.store.service.Versioned;
  32 import org.slf4j.Logger;
  33
  34 import com.fasterxml.jackson.core.JsonGenerator;
  35 import com.fasterxml.jackson.core.JsonParser;
  36 import com.fasterxml.jackson.core.JsonProcessingException;
  37 import com.fasterxml.jackson.databind.DeserializationContext;
  38 import com.fasterxml.jackson.databind.JsonDeserializer;
  39 import com.fasterxml.jackson.databind.JsonNode;
  40 import com.fasterxml.jackson.databind.JsonSerializer;
  41 import com.fasterxml.jackson.databind.ObjectMapper;
  42 import com.fasterxml.jackson.databind.SerializerProvider;
  43 import com.fasterxml.jackson.databind.module.SimpleModule;
  44 import com.google.common.base.Throwables;
  45 import com.google.common.collect.Lists;
  46 import com.google.common.io.Files;
  47
  48 /**
  49  * ClusterMetadataStore backed by a local file.
  50  */
  51 @Component(immediate = true, enabled = true)
  52 @Service
  53 public class StaticClusterMetadataStore
  54     extends AbstractStore<ClusterMetadataEvent, ClusterMetadataStoreDelegate>
  55     implements ClusterMetadataStore {
  56
  57     private final Logger log = getLogger(getClass());
  58     private static final String CLUSTER_METADATA_FILE = "../config/cluster.json";
  59     private static final int DEFAULT_ONOS_PORT = 9876;
  60     private final File metadataFile = new File(CLUSTER_METADATA_FILE);
  61     private AtomicReference<ClusterMetadata> metadata = new AtomicReference<>();
  62     private ObjectMapper mapper;
  63     private long version;
  64
  65     @Activate
  66     public void activate() {
  67         mapper = new ObjectMapper();
  68         SimpleModule module = new SimpleModule();
  69         module.addSerializer(NodeId.class, new NodeIdSerializer());
  70         module.addDeserializer(NodeId.class, new NodeIdDeserializer());
  71         module.addSerializer(ControllerNode.class, new ControllerNodeSerializer());
  72         module.addDeserializer(ControllerNode.class, new ControllerNodeDeserializer());
  73         mapper.registerModule(module);
  74         File metadataFile = new File(CLUSTER_METADATA_FILE);
  75         if (metadataFile.exists()) {
  76             try {
  77                 metadata.set(mapper.readValue(metadataFile, ClusterMetadata.class));
  78                 version = metadataFile.lastModified();
  79             } catch (IOException e) {
  80                 Throwables.propagate(e);
  81             }
  82         } else {
  83             String localIp = getSiteLocalAddress();
  84             ControllerNode localNode =
  85                     new DefaultControllerNode(new NodeId(localIp), IpAddress.valueOf(localIp), DEFAULT_ONOS_PORT);
  86             metadata.set(ClusterMetadata.builder()
  87                     .withName("default")
  88                     .withControllerNodes(Arrays.asList(localNode))
  89                     .withPartitions(Lists.newArrayList(new Partition("p1", Lists.newArrayList(localNode.id()))))
  90                     .build());
  91             version = System.currentTimeMillis();
  92         }
  93         log.info("Started");
  94     }
  95
  96     @Deactivate
  97     public void deactivate() {
  98         log.info("Stopped");
  99     }
 100
 101     @Override
 102     public void setDelegate(ClusterMetadataStoreDelegate delegate) {
 103         checkNotNull(delegate, "Delegate cannot be null");
 104         this.delegate = delegate;
 105     }
 106
 107     @Override
 108     public void unsetDelegate(ClusterMetadataStoreDelegate delegate) {
 109         this.delegate = null;
 110     }
 111
 112     @Override
 113     public boolean hasDelegate() {
 114         return this.delegate != null;
 115     }
 116
 117     @Override
 118     public Versioned<ClusterMetadata> getClusterMetadata() {
 119         return new Versioned<>(metadata.get(), version);
 120     }
 121
 122     @Override
 123     public void setClusterMetadata(ClusterMetadata metadata) {
 124         checkNotNull(metadata);
 125         try {
 126             Files.createParentDirs(metadataFile);
 127             mapper.writeValue(metadataFile, metadata);
 128             this.metadata.set(metadata);
 129         } catch (IOException e) {
 130             Throwables.propagate(e);
 131         }
 132     }
 133
 134     @Override
 135     public void setActiveReplica(String partitionId, NodeId nodeId) {
 136         throw new UnsupportedOperationException();
 137     }
 138
 139     @Override
 140     public void unsetActiveReplica(String partitionId, NodeId nodeId) {
 141         throw new UnsupportedOperationException();
 142     }
 143
 144     @Override
 145     public Collection<NodeId> getActiveReplicas(String partitionId) {
 146         return metadata.get().getPartitions()
 147                        .stream()
 148                        .filter(r -> r.getName().equals(partitionId))
 149                        .findFirst()
 150                        .map(r -> r.getMembers())
 151                        .orElse(null);
 152     }
 153
 154     private static class ControllerNodeSerializer extends JsonSerializer<ControllerNode> {
 155         @Override
 156         public void serialize(ControllerNode node, JsonGenerator jgen, SerializerProvider provider)
 157           throws IOException, JsonProcessingException {
 158             jgen.writeStartObject();
 159             jgen.writeStringField("id", node.id().toString());
 160             jgen.writeStringField("ip", node.ip().toString());
 161             jgen.writeNumberField("port", node.tcpPort());
 162             jgen.writeEndObject();
 163         }
 164     }
 165
 166     private static class ControllerNodeDeserializer extends JsonDeserializer<ControllerNode> {
 167         @Override
 168         public ControllerNode deserialize(JsonParser jp, DeserializationContext ctxt)
 169                 throws IOException, JsonProcessingException {
 170             JsonNode node = jp.getCodec().readTree(jp);
 171             NodeId nodeId = new NodeId(node.get("id").textValue());
 172             IpAddress ip = IpAddress.valueOf(node.get("ip").textValue());
 173             int port = node.get("port").asInt();
 174             return new DefaultControllerNode(nodeId, ip, port);
 175         }
 176     }
 177
 178     private static class NodeIdSerializer extends JsonSerializer<NodeId> {
 179         @Override
 180         public void serialize(NodeId nodeId, JsonGenerator jgen, SerializerProvider provider)
 181           throws IOException, JsonProcessingException {
 182             jgen.writeString(nodeId.toString());
 183         }
 184     }
 185
 186     private class NodeIdDeserializer extends JsonDeserializer<NodeId> {
 187         @Override
 188         public NodeId deserialize(JsonParser jp, DeserializationContext ctxt)
 189           throws IOException, JsonProcessingException {
 190             JsonNode node = jp.getCodec().readTree(jp);
 191             return new NodeId(node.asText());
 192         }
 193     }
 194
 195
 196     private static String getSiteLocalAddress() {
 197         Function<NetworkInterface, IpAddress> ipLookup = nif -> {
 198             for (InetAddress address : Collections.list(nif.getInetAddresses())) {
 199                 if (address.isSiteLocalAddress()) {
 200                     return IpAddress.valueOf(address);
 201                 }
 202             }
 203             return null;
 204         };
 205         try {
 206             IpAddress ip = ipLookup.apply(NetworkInterface.getByName("eth0"));
 207             if (ip != null) {
 208                 return ip.toString();
 209             }
 210             for (NetworkInterface nif : Collections.list(getNetworkInterfaces())) {
 211                 ip = ipLookup.apply(nif);
 212                 if (ip != null) {
 213                     return ip.toString();
 214                 }
 215             }
 216         } catch (Exception e) {
 217             throw new IllegalStateException("Unable to get network interfaces", e);
 218         }
 219         return IpAddress.valueOf(InetAddress.getLoopbackAddress()).toString();
 220     }
 221 }