hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java - hadoop - Git at Google

 /**
  * 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.hadoop.hdds.scm.pipeline;

 import com.google.common.base.Preconditions;
 import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;

 /**
  * Represents a group of datanodes which store a container.
  */
 public final class Pipeline {

   private static final Logger LOG = LoggerFactory.getLogger(Pipeline.class);
   private final PipelineID id;
   private final ReplicationType type;
   private final ReplicationFactor factor;

   private PipelineState state;
   private Map<DatanodeDetails, Long> nodeStatus;
   // nodes with ordered distance to client
   private ThreadLocal<List<DatanodeDetails>> nodesInOrder = new ThreadLocal<>();

   /**
    * The immutable properties of pipeline object is used in
    * ContainerStateManager#getMatchingContainerByPipeline to take a lock on
    * the container allocations for a particular pipeline.
    */
   private Pipeline(PipelineID id, ReplicationType type,
       ReplicationFactor factor, PipelineState state,
       Map<DatanodeDetails, Long> nodeStatus) {
     this.id = id;
     this.type = type;
     this.factor = factor;
     this.state = state;
     this.nodeStatus = nodeStatus;
   }

   /**
    * Returns the ID of this pipeline.
    *
    * @return PipelineID
    */
   public PipelineID getId() {
     return id;
   }

   /**
    * Returns the type.
    *
    * @return type - Simple or Ratis.
    */
   public ReplicationType getType() {
     return type;
   }

   /**
    * Returns the factor.
    *
    * @return type - Simple or Ratis.
    */
   public ReplicationFactor getFactor() {
     return factor;
   }

   /**
    * Returns the State of the pipeline.
    *
    * @return - LifeCycleStates.
    */
   public PipelineState getPipelineState() {
     return state;
   }

   /**
    * Returns the list of nodes which form this pipeline.
    *
    * @return List of DatanodeDetails
    */
   public List<DatanodeDetails> getNodes() {
     return new ArrayList<>(nodeStatus.keySet());
   }

   public DatanodeDetails getFirstNode() throws IOException {
     if (nodeStatus.isEmpty()) {
       throw new IOException(String.format("Pipeline=%s is empty", id));
     }
     return nodeStatus.keySet().iterator().next();
   }

   public DatanodeDetails getClosestNode() throws IOException {
     if (nodesInOrder.get() == null || nodesInOrder.get().isEmpty()) {
       LOG.debug("Nodes in order is empty, delegate to getFirstNode");
       return getFirstNode();
     }
     return nodesInOrder.get().get(0);
   }

   public boolean isClosed() {
     return state == PipelineState.CLOSED;
   }

   public boolean isOpen() {
     return state == PipelineState.OPEN;
   }

   public void setNodesInOrder(List<DatanodeDetails> nodes) {
     nodesInOrder.set(nodes);
   }

   public List<DatanodeDetails> getNodesInOrder() {
     if (nodesInOrder.get() == null || nodesInOrder.get().isEmpty()) {
       LOG.debug("Nodes in order is empty, delegate to getNodes");
       return getNodes();
     }
     return nodesInOrder.get();
   }

   void reportDatanode(DatanodeDetails dn) throws IOException {
     if (nodeStatus.get(dn) == null) {
       throw new IOException(
           String.format("Datanode=%s not part of pipeline=%s", dn, id));
     }
     nodeStatus.put(dn, System.currentTimeMillis());
   }

   boolean isHealthy() {
     for (Long reportedTime : nodeStatus.values()) {
       if (reportedTime < 0) {
         return false;
       }
     }
     return true;
   }

   public boolean isEmpty() {
     return nodeStatus.isEmpty();
   }

   public HddsProtos.Pipeline getProtobufMessage()
       throws UnknownPipelineStateException {
     HddsProtos.Pipeline.Builder builder = HddsProtos.Pipeline.newBuilder()
         .setId(id.getProtobuf())
         .setType(type)
         .setFactor(factor)
         .setState(PipelineState.getProtobuf(state))
         .setLeaderID("")
         .addAllMembers(nodeStatus.keySet().stream()
             .map(DatanodeDetails::getProtoBufMessage)
             .collect(Collectors.toList()));
     // To save the message size on wire, only transfer the node order based on
     // network topology
     List<DatanodeDetails> nodes = nodesInOrder.get();
     if (nodes != null && !nodes.isEmpty()) {
       for (int i = 0; i < nodes.size(); i++) {
         Iterator<DatanodeDetails> it = nodeStatus.keySet().iterator();
         for (int j = 0; j < nodeStatus.keySet().size(); j++) {
           if (it.next().equals(nodes.get(i))) {
             builder.addMemberOrders(j);
             break;
           }
         }
       }
       if (LOG.isDebugEnabled()) {
         LOG.debug("Serialize pipeline {} with nodesInOrder{ }", id.toString(),
             nodes);
       }
     }
     return builder.build();
   }

   public static Pipeline getFromProtobuf(HddsProtos.Pipeline pipeline)
       throws UnknownPipelineStateException {
     Preconditions.checkNotNull(pipeline, "Pipeline is null");
     return new Builder().setId(PipelineID.getFromProtobuf(pipeline.getId()))
         .setFactor(pipeline.getFactor())
         .setType(pipeline.getType())
         .setState(PipelineState.fromProtobuf(pipeline.getState()))
         .setNodes(pipeline.getMembersList().stream()
             .map(DatanodeDetails::getFromProtoBuf).collect(Collectors.toList()))
         .setNodesInOrder(pipeline.getMemberOrdersList())
         .build();
   }

   @Override
   public boolean equals(Object o) {
     if (this == o) {
       return true;
     }
     if (o == null || getClass() != o.getClass()) {
       return false;
     }

     Pipeline that = (Pipeline) o;

     return new EqualsBuilder()
         .append(id, that.id)
         .append(type, that.type)
         .append(factor, that.factor)
         .append(getNodes(), that.getNodes())
         .isEquals();
   }

   @Override
   public int hashCode() {
     return new HashCodeBuilder()
         .append(id)
         .append(type)
         .append(factor)
         .append(nodeStatus)
         .toHashCode();
   }

   @Override
   public String toString() {
     final StringBuilder b =
         new StringBuilder(getClass().getSimpleName()).append("[");
     b.append(" Id: ").append(id.getId());
     b.append(", Nodes: ");
     nodeStatus.keySet().forEach(b::append);
     b.append(", Type:").append(getType());
     b.append(", Factor:").append(getFactor());
     b.append(", State:").append(getPipelineState());
     b.append("]");
     return b.toString();
   }

   public static Builder newBuilder() {
     return new Builder();
   }

   public static Builder newBuilder(Pipeline pipeline) {
     return new Builder(pipeline);
   }

   /**
    * Builder class for Pipeline.
    */
   public static class Builder {
     private PipelineID id = null;
     private ReplicationType type = null;
     private ReplicationFactor factor = null;
     private PipelineState state = null;
     private Map<DatanodeDetails, Long> nodeStatus = null;
     private List<Integer> nodeOrder = null;
     private List<DatanodeDetails> nodesInOrder = null;

     public Builder() {}

     public Builder(Pipeline pipeline) {
       this.id = pipeline.id;
       this.type = pipeline.type;
       this.factor = pipeline.factor;
       this.state = pipeline.state;
       this.nodeStatus = pipeline.nodeStatus;
       this.nodesInOrder = pipeline.nodesInOrder.get();
     }

     public Builder setId(PipelineID id1) {
       this.id = id1;
       return this;
     }

     public Builder setType(ReplicationType type1) {
       this.type = type1;
       return this;
     }

     public Builder setFactor(ReplicationFactor factor1) {
       this.factor = factor1;
       return this;
     }

     public Builder setState(PipelineState state1) {
       this.state = state1;
       return this;
     }

     public Builder setNodes(List<DatanodeDetails> nodes) {
       this.nodeStatus = new LinkedHashMap<>();
       nodes.forEach(node -> nodeStatus.put(node, -1L));
       return this;
     }

     public Builder setNodesInOrder(List<Integer> orders) {
       this.nodeOrder = orders;
       return this;
     }

     public Pipeline build() {
       Preconditions.checkNotNull(id);
       Preconditions.checkNotNull(type);
       Preconditions.checkNotNull(factor);
       Preconditions.checkNotNull(state);
       Preconditions.checkNotNull(nodeStatus);
       Pipeline pipeline = new Pipeline(id, type, factor, state, nodeStatus);

       if (nodeOrder != null && !nodeOrder.isEmpty()) {
         // This branch is for build from ProtoBuf
         List<DatanodeDetails> nodesWithOrder = new ArrayList<>();
         for(int i = 0; i < nodeOrder.size(); i++) {
           int nodeIndex = nodeOrder.get(i);
           Iterator<DatanodeDetails> it = nodeStatus.keySet().iterator();
           while(it.hasNext() && nodeIndex >= 0) {
             DatanodeDetails node = it.next();
             if (nodeIndex == 0) {
               nodesWithOrder.add(node);
               break;
             }
             nodeIndex--;
           }
         }
         if (LOG.isDebugEnabled()) {
           LOG.debug("Deserialize nodesInOrder {} in pipeline {}",
               nodesWithOrder, id.toString());
         }
         pipeline.setNodesInOrder(nodesWithOrder);
       } else if (nodesInOrder != null){
         // This branch is for pipeline clone
         pipeline.setNodesInOrder(nodesInOrder);
       }
       return pipeline;
     }
   }

   /**
    * Possible Pipeline states in SCM.
    */
   public enum PipelineState {
     ALLOCATED, OPEN, DORMANT, CLOSED;

     public static PipelineState fromProtobuf(HddsProtos.PipelineState state)
         throws UnknownPipelineStateException {
       Preconditions.checkNotNull(state, "Pipeline state is null");
       switch (state) {
       case PIPELINE_ALLOCATED: return ALLOCATED;
       case PIPELINE_OPEN: return OPEN;
       case PIPELINE_DORMANT: return DORMANT;
       case PIPELINE_CLOSED: return CLOSED;
       default:
         throw new UnknownPipelineStateException(
             "Pipeline state: " + state + " is not recognized.");
       }
     }

     public static HddsProtos.PipelineState getProtobuf(PipelineState state)
         throws UnknownPipelineStateException {
       Preconditions.checkNotNull(state, "Pipeline state is null");
       switch (state) {
       case ALLOCATED: return HddsProtos.PipelineState.PIPELINE_ALLOCATED;
       case OPEN: return HddsProtos.PipelineState.PIPELINE_OPEN;
       case DORMANT: return HddsProtos.PipelineState.PIPELINE_DORMANT;
       case CLOSED: return HddsProtos.PipelineState.PIPELINE_CLOSED;
       default:
         throw new UnknownPipelineStateException(
             "Pipeline state: " + state + " is not recognized.");
       }
     }
   }
 }
	/**
	* 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.hadoop.hdds.scm.pipeline;

	import com.google.common.base.Preconditions;
	import org.apache.commons.lang3.builder.EqualsBuilder;
	import org.apache.commons.lang3.builder.HashCodeBuilder;
	import org.apache.hadoop.hdds.protocol.DatanodeDetails;
	import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
	import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
	import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.Iterator;
	import java.util.LinkedHashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.stream.Collectors;

	/**
	* Represents a group of datanodes which store a container.
	*/
	public final class Pipeline {

	private static final Logger LOG = LoggerFactory.getLogger(Pipeline.class);
	private final PipelineID id;
	private final ReplicationType type;
	private final ReplicationFactor factor;

	private PipelineState state;
	private Map<DatanodeDetails, Long> nodeStatus;
	// nodes with ordered distance to client
	private ThreadLocal<List<DatanodeDetails>> nodesInOrder = new ThreadLocal<>();

	/**
	* The immutable properties of pipeline object is used in
	* ContainerStateManager#getMatchingContainerByPipeline to take a lock on
	* the container allocations for a particular pipeline.
	*/
	private Pipeline(PipelineID id, ReplicationType type,
	ReplicationFactor factor, PipelineState state,
	Map<DatanodeDetails, Long> nodeStatus) {
	this.id = id;
	this.type = type;
	this.factor = factor;
	this.state = state;
	this.nodeStatus = nodeStatus;
	}

	/**
	* Returns the ID of this pipeline.
	*
	* @return PipelineID
	*/
	public PipelineID getId() {
	return id;
	}

	/**
	* Returns the type.
	*
	* @return type - Simple or Ratis.
	*/
	public ReplicationType getType() {
	return type;
	}

	/**
	* Returns the factor.
	*
	* @return type - Simple or Ratis.
	*/
	public ReplicationFactor getFactor() {
	return factor;
	}

	/**
	* Returns the State of the pipeline.
	*
	* @return - LifeCycleStates.
	*/
	public PipelineState getPipelineState() {
	return state;
	}

	/**
	* Returns the list of nodes which form this pipeline.
	*
	* @return List of DatanodeDetails
	*/
	public List<DatanodeDetails> getNodes() {
	return new ArrayList<>(nodeStatus.keySet());
	}

	public DatanodeDetails getFirstNode() throws IOException {
	if (nodeStatus.isEmpty()) {
	throw new IOException(String.format("Pipeline=%s is empty", id));
	}
	return nodeStatus.keySet().iterator().next();
	}

	public DatanodeDetails getClosestNode() throws IOException {
	if (nodesInOrder.get() == null \|\| nodesInOrder.get().isEmpty()) {
	LOG.debug("Nodes in order is empty, delegate to getFirstNode");
	return getFirstNode();
	}
	return nodesInOrder.get().get(0);
	}

	public boolean isClosed() {
	return state == PipelineState.CLOSED;
	}

	public boolean isOpen() {
	return state == PipelineState.OPEN;
	}

	public void setNodesInOrder(List<DatanodeDetails> nodes) {
	nodesInOrder.set(nodes);
	}

	public List<DatanodeDetails> getNodesInOrder() {
	if (nodesInOrder.get() == null \|\| nodesInOrder.get().isEmpty()) {
	LOG.debug("Nodes in order is empty, delegate to getNodes");
	return getNodes();
	}
	return nodesInOrder.get();
	}

	void reportDatanode(DatanodeDetails dn) throws IOException {
	if (nodeStatus.get(dn) == null) {
	throw new IOException(
	String.format("Datanode=%s not part of pipeline=%s", dn, id));
	}
	nodeStatus.put(dn, System.currentTimeMillis());
	}

	boolean isHealthy() {
	for (Long reportedTime : nodeStatus.values()) {
	if (reportedTime < 0) {
	return false;
	}
	}
	return true;
	}

	public boolean isEmpty() {
	return nodeStatus.isEmpty();
	}

	public HddsProtos.Pipeline getProtobufMessage()
	throws UnknownPipelineStateException {
	HddsProtos.Pipeline.Builder builder = HddsProtos.Pipeline.newBuilder()
	.setId(id.getProtobuf())
	.setType(type)
	.setFactor(factor)
	.setState(PipelineState.getProtobuf(state))
	.setLeaderID("")
	.addAllMembers(nodeStatus.keySet().stream()
	.map(DatanodeDetails::getProtoBufMessage)
	.collect(Collectors.toList()));
	// To save the message size on wire, only transfer the node order based on
	// network topology
	List<DatanodeDetails> nodes = nodesInOrder.get();
	if (nodes != null && !nodes.isEmpty()) {
	for (int i = 0; i < nodes.size(); i++) {
	Iterator<DatanodeDetails> it = nodeStatus.keySet().iterator();
	for (int j = 0; j < nodeStatus.keySet().size(); j++) {
	if (it.next().equals(nodes.get(i))) {
	builder.addMemberOrders(j);
	break;
	}
	}
	}
	if (LOG.isDebugEnabled()) {
	LOG.debug("Serialize pipeline {} with nodesInOrder{ }", id.toString(),
	nodes);
	}
	}
	return builder.build();
	}

	public static Pipeline getFromProtobuf(HddsProtos.Pipeline pipeline)
	throws UnknownPipelineStateException {
	Preconditions.checkNotNull(pipeline, "Pipeline is null");
	return new Builder().setId(PipelineID.getFromProtobuf(pipeline.getId()))
	.setFactor(pipeline.getFactor())
	.setType(pipeline.getType())
	.setState(PipelineState.fromProtobuf(pipeline.getState()))
	.setNodes(pipeline.getMembersList().stream()
	.map(DatanodeDetails::getFromProtoBuf).collect(Collectors.toList()))
	.setNodesInOrder(pipeline.getMemberOrdersList())
	.build();
	}

	@Override
	public boolean equals(Object o) {
	if (this == o) {
	return true;
	}
	if (o == null \|\| getClass() != o.getClass()) {
	return false;
	}

	Pipeline that = (Pipeline) o;

	return new EqualsBuilder()
	.append(id, that.id)
	.append(type, that.type)
	.append(factor, that.factor)
	.append(getNodes(), that.getNodes())
	.isEquals();
	}

	@Override
	public int hashCode() {
	return new HashCodeBuilder()
	.append(id)
	.append(type)
	.append(factor)
	.append(nodeStatus)
	.toHashCode();
	}

	@Override
	public String toString() {
	final StringBuilder b =
	new StringBuilder(getClass().getSimpleName()).append("[");
	b.append(" Id: ").append(id.getId());
	b.append(", Nodes: ");
	nodeStatus.keySet().forEach(b::append);
	b.append(", Type:").append(getType());
	b.append(", Factor:").append(getFactor());
	b.append(", State:").append(getPipelineState());
	b.append("]");
	return b.toString();
	}

	public static Builder newBuilder() {
	return new Builder();
	}

	public static Builder newBuilder(Pipeline pipeline) {
	return new Builder(pipeline);
	}

	/**
	* Builder class for Pipeline.
	*/
	public static class Builder {
	private PipelineID id = null;
	private ReplicationType type = null;
	private ReplicationFactor factor = null;
	private PipelineState state = null;
	private Map<DatanodeDetails, Long> nodeStatus = null;
	private List<Integer> nodeOrder = null;
	private List<DatanodeDetails> nodesInOrder = null;

	public Builder() {}

	public Builder(Pipeline pipeline) {
	this.id = pipeline.id;
	this.type = pipeline.type;
	this.factor = pipeline.factor;
	this.state = pipeline.state;
	this.nodeStatus = pipeline.nodeStatus;
	this.nodesInOrder = pipeline.nodesInOrder.get();
	}

	public Builder setId(PipelineID id1) {
	this.id = id1;
	return this;
	}

	public Builder setType(ReplicationType type1) {
	this.type = type1;
	return this;
	}

	public Builder setFactor(ReplicationFactor factor1) {
	this.factor = factor1;
	return this;
	}

	public Builder setState(PipelineState state1) {
	this.state = state1;
	return this;
	}

	public Builder setNodes(List<DatanodeDetails> nodes) {
	this.nodeStatus = new LinkedHashMap<>();
	nodes.forEach(node -> nodeStatus.put(node, -1L));
	return this;
	}

	public Builder setNodesInOrder(List<Integer> orders) {
	this.nodeOrder = orders;
	return this;
	}

	public Pipeline build() {
	Preconditions.checkNotNull(id);
	Preconditions.checkNotNull(type);
	Preconditions.checkNotNull(factor);
	Preconditions.checkNotNull(state);
	Preconditions.checkNotNull(nodeStatus);
	Pipeline pipeline = new Pipeline(id, type, factor, state, nodeStatus);

	if (nodeOrder != null && !nodeOrder.isEmpty()) {
	// This branch is for build from ProtoBuf
	List<DatanodeDetails> nodesWithOrder = new ArrayList<>();
	for(int i = 0; i < nodeOrder.size(); i++) {
	int nodeIndex = nodeOrder.get(i);
	Iterator<DatanodeDetails> it = nodeStatus.keySet().iterator();
	while(it.hasNext() && nodeIndex >= 0) {
	DatanodeDetails node = it.next();
	if (nodeIndex == 0) {
	nodesWithOrder.add(node);
	break;
	}
	nodeIndex--;
	}
	}
	if (LOG.isDebugEnabled()) {
	LOG.debug("Deserialize nodesInOrder {} in pipeline {}",
	nodesWithOrder, id.toString());
	}
	pipeline.setNodesInOrder(nodesWithOrder);
	} else if (nodesInOrder != null){
	// This branch is for pipeline clone
	pipeline.setNodesInOrder(nodesInOrder);
	}
	return pipeline;
	}
	}

	/**
	* Possible Pipeline states in SCM.
	*/
	public enum PipelineState {
	ALLOCATED, OPEN, DORMANT, CLOSED;

	public static PipelineState fromProtobuf(HddsProtos.PipelineState state)
	throws UnknownPipelineStateException {
	Preconditions.checkNotNull(state, "Pipeline state is null");
	switch (state) {
	case PIPELINE_ALLOCATED: return ALLOCATED;
	case PIPELINE_OPEN: return OPEN;
	case PIPELINE_DORMANT: return DORMANT;
	case PIPELINE_CLOSED: return CLOSED;
	default:
	throw new UnknownPipelineStateException(
	"Pipeline state: " + state + " is not recognized.");
	}
	}

	public static HddsProtos.PipelineState getProtobuf(PipelineState state)
	throws UnknownPipelineStateException {
	Preconditions.checkNotNull(state, "Pipeline state is null");
	switch (state) {
	case ALLOCATED: return HddsProtos.PipelineState.PIPELINE_ALLOCATED;
	case OPEN: return HddsProtos.PipelineState.PIPELINE_OPEN;
	case DORMANT: return HddsProtos.PipelineState.PIPELINE_DORMANT;
	case CLOSED: return HddsProtos.PipelineState.PIPELINE_CLOSED;
	default:
	throw new UnknownPipelineStateException(
	"Pipeline state: " + state + " is not recognized.");
	}
	}
	}
	}