Structure shard status as tree

pkg/apis/replication/v1alpha/collection_status.go

@@ -0,0 +1,32 @@
+//
+// DISCLAIMER
+//
+// Copyright 2018 ArangoDB GmbH, Cologne, Germany
+//
+// 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.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+// Author Ewout Prangsma
+//
+
+package v1alpha
+
+// CollectionStatus contains the status of a single collection.
+type CollectionStatus struct {
+	// Name of the collection
+	Name string `json:"name"`
+	// Replication status per shard.
+	// The list is ordered by shard index (0..noShards-1)
+	Shards []ShardStatus `json:"shards,omitempty"`
+}

pkg/apis/replication/v1alpha/database_status.go

@@ -0,0 +1,32 @@
+//
+// DISCLAIMER
+//
+// Copyright 2018 ArangoDB GmbH, Cologne, Germany
+//
+// 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.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+// Author Ewout Prangsma
+//
+
+package v1alpha
+
+// DatabaseStatus contains the status of a single database.
+type DatabaseStatus struct {
+	// Name of the database
+	Name string `json:"name"`
+	// Collections holds the replication status of each collection in the database.
+	// List is ordered by name of the collection.
+	Collections []CollectionStatus `json:"collections,omitempty"`
+}

pkg/apis/replication/v1alpha/endpoint_status.go

@@ -24,5 +24,7 @@ package v1alpha
 
 // EndpointStatus contains the status of either the source or destination endpoint.
 type EndpointStatus struct {
-	Shards []ShardStatus `json:"shards,omitempty"`
+	// Databases holds the replication status of all databases from the point of view of this endpoint.
+	// List is ordered by name of the database.
+	Databases []DatabaseStatus `json:"databases,omitempty"`
 }

pkg/apis/replication/v1alpha/replication_status.go

@@ -26,7 +26,7 @@ package v1alpha
 // an ArangoDeploymentReplication.
 type DeploymentReplicationStatus struct {
 	// Phase holds the current lifetime phase of the deployment replication
-	Phase DeploymentReplicationPhase `json:"phase"`
+	Phase DeploymentReplicationPhase `json:"phase,omitempty"`
 	// Reason contains a human readable reason for reaching the current phase (can be empty)
 	Reason string `json:"reason,omitempty"` // Reason for current phase
 

pkg/apis/replication/v1alpha/shard_status.go

@@ -24,8 +24,5 @@ package v1alpha
 
 // ShardStatus contains the status of a single shard.
 type ShardStatus struct {
-	Database   string `json:"database"`
-	Collection string `json:"collection"`
-	ShardIndex int    `json:"shard"`
-	Status     string `json:"status"`
+	Status string `json:"status"`
 }

pkg/apis/replication/v1alpha/zz_generated.deepcopy.go

@@ -89,6 +89,27 @@ func (in *ArangoDeploymentReplicationList) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *CollectionStatus) DeepCopyInto(out *CollectionStatus) {
+	*out = *in
+	if in.Shards != nil {
+		in, out := &in.Shards, &out.Shards
+		*out = make([]ShardStatus, len(*in))
+		copy(*out, *in)
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CollectionStatus.
+func (in *CollectionStatus) DeepCopy() *CollectionStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(CollectionStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Condition) DeepCopyInto(out *Condition) {
 	*out = *in
@@ -107,6 +128,29 @@ func (in *Condition) DeepCopy() *Condition {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *DatabaseStatus) DeepCopyInto(out *DatabaseStatus) {
+	*out = *in
+	if in.Collections != nil {
+		in, out := &in.Collections, &out.Collections
+		*out = make([]CollectionStatus, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DatabaseStatus.
+func (in *DatabaseStatus) DeepCopy() *DatabaseStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(DatabaseStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *DeploymentReplicationSpec) DeepCopyInto(out *DeploymentReplicationSpec) {
 	*out = *in
@@ -219,10 +263,12 @@ func (in *EndpointSpec) DeepCopy() *EndpointSpec {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *EndpointStatus) DeepCopyInto(out *EndpointStatus) {
 	*out = *in
-	if in.Shards != nil {
-		in, out := &in.Shards, &out.Shards
-		*out = make([]ShardStatus, len(*in))
-		copy(*out, *in)
+	if in.Databases != nil {
+		in, out := &in.Databases, &out.Databases
+		*out = make([]DatabaseStatus, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
 	}
 	return
 }

pkg/replication/sync_inspector.go

@@ -24,6 +24,7 @@ package replication
 
 import (
 	"context"
+	"sort"
 	"time"
 
 	"github.com/arangodb/arangosync/client"
@@ -240,28 +241,66 @@ func (dr *DeploymentReplication) hasOutgoingEndpoint(status client.SyncInfo, epS
 func createEndpointStatus(status client.SyncInfo, outgoingID string) api.EndpointStatus {
 	result := api.EndpointStatus{}
 	if outgoingID == "" {
-		for _, s := range status.Shards {
-			result.Shards = append(result.Shards, api.ShardStatus{
-				Database:   s.Database,
-				Collection: s.Collection,
-				ShardIndex: s.ShardIndex,
-				Status:     string(s.Status),
-			})
+		return createEndpointStatusFromShards(status.Shards)
+	}
+	for _, o := range status.Outgoing {
+		if o.ID != outgoingID {
+			continue
 		}
-	} else {
-		for _, o := range status.Outgoing {
-			if o.ID != outgoingID {
-				continue
-			}
-			for _, s := range o.Shards {
-				result.Shards = append(result.Shards, api.ShardStatus{
-					Database:   s.Database,
-					Collection: s.Collection,
-					ShardIndex: s.ShardIndex,
-					Status:     string(s.Status),
-				})
+		return createEndpointStatusFromShards(o.Shards)
+	}
+
+	return result
+}
+
+// createEndpointStatusFromShards creates an api EndpointStatus from the given list of shard statuses.
+func createEndpointStatusFromShards(shards []client.ShardSyncInfo) api.EndpointStatus {
+	result := api.EndpointStatus{}
+
+	getDatabase := func(name string) *api.DatabaseStatus {
+		for i, d := range result.Databases {
+			if d.Name == name {
+				return &result.Databases[i]
 			}
 		}
+		// Not found, add it
+		result.Databases = append(result.Databases, api.DatabaseStatus{Name: name})
+		return &result.Databases[len(result.Databases)-1]
+	}
+
+	getCollection := func(db *api.DatabaseStatus, name string) *api.CollectionStatus {
+		for i, c := range db.Collections {
+			if c.Name == name {
+				return &db.Collections[i]
+			}
+		}
+		// Not found, add it
+		db.Collections = append(db.Collections, api.CollectionStatus{Name: name})
+		return &db.Collections[len(db.Collections)-1]
+	}
+
+	// Sort shard by index
+	sort.Slice(shards, func(i, j int) bool {
+		return shards[i].ShardIndex < shards[j].ShardIndex
+	})
+	for _, s := range shards {
+		db := getDatabase(s.Database)
+		col := getCollection(db, s.Collection)
+
+		// Add "missing" shards if needed
+		for len(col.Shards) < s.ShardIndex {
+			col.Shards = append(col.Shards, api.ShardStatus{Status: ""})
+		}
+
+		// Add current shard
+		col.Shards = append(col.Shards, api.ShardStatus{Status: string(s.Status)})
+	}
+
+	// Sort result
+	sort.Slice(result.Databases, func(i, j int) bool { return result.Databases[i].Name < result.Databases[j].Name })
+	for i, db := range result.Databases {
+		sort.Slice(db.Collections, func(i, j int) bool { return db.Collections[i].Name < db.Collections[j].Name })
+		result.Databases[i] = db
 	}
 	return result
 }