enable topology support

Signed-off-by: Mucahit Kurt <[email protected]>

Author: mucahitkurt

pkg/hostpath/controllerserver.go

@@ -49,12 +49,13 @@ const (
 )
 
 type controllerServer struct {
-	caps []*csi.ControllerServiceCapability
+	caps   []*csi.ControllerServiceCapability
+	nodeID string
 }
 
-func NewControllerServer(ephemeral bool) *controllerServer {
+func NewControllerServer(ephemeral bool, nodeID string) *controllerServer {
 	if ephemeral {
-		return &controllerServer{caps: getControllerServiceCapabilities(nil)}
+		return &controllerServer{caps: getControllerServiceCapabilities(nil), nodeID: nodeID}
 	}
 	return &controllerServer{
 		caps: getControllerServiceCapabilities(
@@ -64,6 +65,7 @@ func NewControllerServer(ephemeral bool) *controllerServer {
 				csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,
 				csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
 			}),
+		nodeID: nodeID,
 	}
 }
 
@@ -164,12 +166,17 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 		glog.V(4).Infof("successfully populated volume %s", vol.VolID)
 	}
 
+	topologies := []*csi.Topology{&csi.Topology{
+		Segments: map[string]string{TopologyKeyNode: cs.nodeID},
+	}}
+
 	return &csi.CreateVolumeResponse{
 		Volume: &csi.Volume{
-			VolumeId:      volumeID,
-			CapacityBytes: req.GetCapacityRange().GetRequiredBytes(),
-			VolumeContext: req.GetParameters(),
-			ContentSource: req.GetVolumeContentSource(),
+			VolumeId:           volumeID,
+			CapacityBytes:      req.GetCapacityRange().GetRequiredBytes(),
+			VolumeContext:      req.GetParameters(),
+			ContentSource:      req.GetVolumeContentSource(),
+			AccessibleTopology: topologies,
 		},
 	}, nil
 }

pkg/hostpath/hostpath.go

@@ -126,7 +126,7 @@ func (hp *hostPath) Run() {
 	// Create GRPC servers
 	hp.ids = NewIdentityServer(hp.name, hp.version)
 	hp.ns = NewNodeServer(hp.nodeID, hp.ephemeral)
-	hp.cs = NewControllerServer(hp.ephemeral)
+	hp.cs = NewControllerServer(hp.ephemeral, hp.nodeID)
 
 	s := NewNonBlockingGRPCServer()
 	s.Start(hp.endpoint, hp.ids, hp.cs, hp.ns)

pkg/hostpath/identityserver.go

@@ -68,6 +68,13 @@ func (ids *identityServer) GetPluginCapabilities(ctx context.Context, req *csi.G
 					},
 				},
 			},
+			{
+				Type: &csi.PluginCapability_Service_{
+					Service: &csi.PluginCapability_Service{
+						Type: csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS,
+					},
+				},
+			},
 		},
 	}, nil
 }

pkg/hostpath/nodeserver.go

@@ -31,6 +31,8 @@ import (
 	"k8s.io/kubernetes/pkg/volume/util/volumepathhandler"
 )
 
+const TopologyKeyNode = "topology.hostpath.csi.k8s.io/node"
+
 type nodeServer struct {
 	nodeID    string
 	ephemeral bool
@@ -261,8 +263,13 @@ func (ns *nodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstag
 
 func (ns *nodeServer) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) (*csi.NodeGetInfoResponse, error) {
 
+	topology := &csi.Topology{
+		Segments: map[string]string{TopologyKeyNode: ns.nodeID},
+	}
+
 	return &csi.NodeGetInfoResponse{
-		NodeId: ns.nodeID,
+		NodeId:             ns.nodeID,
+		AccessibleTopology: topology,
 	}, nil
 }