Expose router feature dataConnectionCount with flag on link create co…

api/types/client.go

@@ -17,6 +17,7 @@ type ConnectorCreateOptions struct {
 	SkupperNamespace string
 	Name             string
 	Cost             int32
+	ConCount         int
 }
 
 type ConnectorRemoveOptions struct {
@@ -29,6 +30,7 @@ type LinkStatus struct {
 	Name        string
 	Url         string
 	Cost        int
+	ConCount    int
 	Connected   bool
 	Configured  bool
 	Description string

api/types/types.go

@@ -168,6 +168,7 @@ const (
 	TokenGeneratedBy            string = BaseQualifier + "/generated-by"
 	SiteVersion                 string = BaseQualifier + "/site-version"
 	TokenCost                   string = BaseQualifier + "/cost"
+	TokenConCount               string = BaseQualifier + "/con-count"
 	UpdatedAnnotation           string = InternalQualifier + "/updated"
 	AnnotationExcludes          string = BaseQualifier + "/exclude-annotations"
 	LabelExcludes               string = BaseQualifier + "/exclude-labels"
@@ -431,6 +432,7 @@ type Connector struct {
 	VerifyHostname bool   `json:"verifyHostname,omitempty"`
 	SslProfile     string `json:"sslProfile,omitempty"`
 	LinkCapacity   int32  `json:"linkCapacity,omitempty"`
+	ConCount       int    `json:"dataConnectionCount"`
 }
 
 type Credential struct {

client/connector_create.go

@@ -167,6 +167,12 @@ func (cli *VanClient) ConnectorCreateSecretFromData(ctx context.Context, secretD
 				}
 				secret.ObjectMeta.Annotations[types.TokenCost] = strconv.Itoa(int(options.Cost))
 			}
+			if options.ConCount != 0 {
+				if secret.ObjectMeta.Annotations == nil {
+					secret.ObjectMeta.Annotations = map[string]string{}
+				}
+				secret.ObjectMeta.Annotations[types.TokenConCount] = strconv.Itoa(int(options.ConCount))
+			}
 			_, err = cli.KubeClient.CoreV1().Secrets(options.SkupperNamespace).Create(&secret)
 			if err == nil {
 				return &secret, nil
@@ -268,6 +274,7 @@ func (cli *VanClient) ConnectorCreate(ctx context.Context, secret *corev1.Secret
 			Name:       options.Name,
 			Cost:       options.Cost,
 			SslProfile: profileName,
+			ConCount:   options.ConCount,
 		}
 		connector.SetMaxFrameSize(siteConfig.Spec.Router.MaxFrameSize)
 		connector.SetMaxSessionFrames(siteConfig.Spec.Router.MaxSessionFrames)

client/connector_list.go

@@ -35,6 +35,7 @@ func getLinkStatus(s *corev1.Secret, edge bool, connections []qdr.Connection) ty
 		if connection := kubeqdr.GetInterRouterOrEdgeConnection(link.Url, connections); connection != nil && connection.Active {
 			link.Connected = true
 			link.Cost, _ = strconv.Atoi(s.ObjectMeta.Annotations[types.TokenCost])
+			link.ConCount, _ = strconv.Atoi(s.ObjectMeta.Annotations[types.TokenConCount])
 			link.Created = s.ObjectMeta.CreationTimestamp.String()
 		}
 		if s.ObjectMeta.Labels[types.SkupperDisabledQualifier] == "true" {

cmd/service-controller/links.go

@@ -94,6 +94,12 @@ func getLinkFromClaim(s *corev1.Secret) *types.LinkStatus {
 				link.Cost = cost
 			}
 		}
+		if value, ok := s.ObjectMeta.Annotations[types.TokenConCount]; ok {
+			conCount, err := strconv.Atoi(value)
+			if err == nil {
+				link.ConCount = conCount
+			}
+		}
 	}
 	return &link
 }

cmd/service-controller/token_handler.go

@@ -62,6 +62,21 @@ func (c *TokenHandler) getTokenCost(token *corev1.Secret) (int32, bool) {
 	return 0, false
 }
 
+func (c *TokenHandler) getTokenConCount(token *corev1.Secret) (int, bool) {
+	if token.ObjectMeta.Annotations == nil {
+		return 0, false
+	}
+	if conCountString, ok := token.ObjectMeta.Annotations[types.TokenConCount]; ok {
+		conCount, err := strconv.Atoi(conCountString)
+		if err != nil {
+			event.Recordf(c.name, "Ignoring invalid conCount annotation %q", conCountString)
+			return 0, false
+		}
+		return int(conCount), true
+	}
+	return 0, false
+}
+
 func (c *TokenHandler) connect(token *corev1.Secret) error {
 	event.Recordf(c.name, "Connecting using token %s", token.ObjectMeta.Name)
 	var options types.ConnectorCreateOptions
@@ -70,6 +85,9 @@ func (c *TokenHandler) connect(token *corev1.Secret) error {
 	if cost, ok := c.getTokenCost(token); ok {
 		options.Cost = cost
 	}
+	if conCount, ok := c.getTokenConCount(token); ok {
+		options.ConCount = conCount
+	}
 	return c.vanClient.ConnectorCreate(context.Background(), token, options)
 }
 

cmd/service-controller/token_handler_test.go

@@ -156,6 +156,53 @@ func TestTokenHandler(t *testing.T) {
 			},
 			expectedConnector: nil,
 		},
+		{ // Set data-connection-count to 3
+			name: "six",
+			annotations: map[string]string{
+				types.TokenConCount: "3",
+				"inter-router-host": "myrouter.com",
+				"inter-router-port": "55671",
+			},
+			expectedConnector: &qdr.Connector{
+				Name:       "six",
+				Host:       "myrouter.com",
+				Port:       "55671",
+				Cost:       0,
+				SslProfile: "six-profile",
+				ConCount:   3,
+			},
+		},
+		{ // Do not set data-connection-count
+			name: "seven",
+			annotations: map[string]string{
+				"inter-router-host": "myrouter.com",
+				"inter-router-port": "55671",
+			},
+			expectedConnector: &qdr.Connector{
+				Name:       "seven",
+				Host:       "myrouter.com",
+				Port:       "55671",
+				Cost:       0,
+				SslProfile: "seven-profile",
+				ConCount:   0,
+			},
+		},
+		{ // Set data-connection-count to a bad value
+			name: "eight",
+			annotations: map[string]string{
+				"inter-router-host": "myrouter.com",
+				"inter-router-port": "55671",
+				types.TokenConCount: "foo",
+			},
+			expectedConnector: &qdr.Connector{
+				Name:       "eight",
+				Host:       "myrouter.com",
+				Port:       "55671",
+				Cost:       0,
+				SslProfile: "eight-profile",
+				ConCount:   0,
+			},
+		},
 	}
 	for _, test := range tests {
 		token := createToken(test.name, test.annotations)
@@ -182,6 +229,7 @@ func TestTokenHandler(t *testing.T) {
 			assert.Equal(t, connector.Host, test.expectedConnector.Host, test.name)
 			assert.Equal(t, connector.Port, test.expectedConnector.Port, test.name)
 			assert.Equal(t, connector.Cost, test.expectedConnector.Cost, test.name)
+			assert.Equal(t, connector.ConCount, test.expectedConnector.ConCount, test.name)
 			assert.Equal(t, connector.SslProfile, test.expectedConnector.SslProfile, test.name)
 
 			//now disconnect:

cmd/site-controller/controller.go

@@ -287,6 +287,21 @@ func getTokenCost(token *corev1.Secret) (int32, bool) {
 	return 0, false
 }
 
+func getTokenConCount(token *corev1.Secret) (int, bool) {
+	if token.ObjectMeta.Annotations == nil {
+		return 0, false
+	}
+	if conCountString, ok := token.ObjectMeta.Annotations[types.TokenConCount]; ok {
+		conCount, err := strconv.Atoi(conCountString)
+		if err != nil {
+			log.Printf("Ignoring invalid conCount annotation %q", conCountString)
+			return 0, false
+		}
+		return int(conCount), true
+	}
+	return 0, false
+}
+
 func (c *SiteController) connect(token *corev1.Secret, namespace string) error {
 	log.Printf("Connecting site in %s using token %s", namespace, token.ObjectMeta.Name)
 	var options types.ConnectorCreateOptions
@@ -295,6 +310,9 @@ func (c *SiteController) connect(token *corev1.Secret, namespace string) error {
 	if cost, ok := getTokenCost(token); ok {
 		options.Cost = cost
 	}
+	if conCount, ok := getTokenConCount(token); ok {
+		options.ConCount = conCount
+	}
 	return c.vanClient.ConnectorCreate(context.Background(), token, options)
 }
 

cmd/site-controller/controller_cluster_test.go

@@ -261,6 +261,7 @@ func TestSiteControlleWithCluster(t *testing.T) {
 		Data: currentToken.Data,
 	}
 	connectSecret.ObjectMeta.Annotations[types.TokenCost] = "5"
+	connectSecret.ObjectMeta.Annotations[types.TokenConCount] = "2"
 
 	os.Setenv("WATCH_NAMESPACE", privateNamespace)
 	privateController, err := NewSiteController(privateCli)

cmd/skupper/skupper_kube_link.go

@@ -36,6 +36,7 @@ func (s *SkupperKubeLink) Create(cmd *cobra.Command, args []string) error {
 		os.Exit(1)
 	}
 	connectorCreateOpts.SkupperNamespace = cli.GetNamespace()
+	connectorCreateOpts.ConCount, _ = cmd.Flags().GetInt("data-connection-count")
 	yaml, err := ioutil.ReadFile(args[0])
 	if err != nil {
 		return fmt.Errorf("Could not read connection token: %s", err.Error())

cmd/skupper/skupper_link.go

@@ -34,6 +34,7 @@ func NewCmdLinkCreate(skupperClient SkupperLinkClient, flag string) *cobra.Comma
 	}
 	cmd.Flags().StringVarP(&connectorCreateOpts.Name, flag, "", "", "Provide a specific name for the link (used when deleting it)")
 	cmd.Flags().Int32VarP(&connectorCreateOpts.Cost, "cost", "", 1, "Specify a cost for this link.")
+	cmd.Flags().IntVarP(&connectorCreateOpts.ConCount, "data-connection-count", "", 0, "How many data links for each inter-router connection. A zero value (default) means no value is set. Router will use its default value.")
 
 	return cmd
 }

pkg/qdr/amqp_mgmt.go

@@ -1165,6 +1165,7 @@ type ConnectorStatus struct {
 	Port        string
 	Role        string
 	Cost        int
+	ConCount    int
 	Status      string
 	Description string
 }
@@ -1176,6 +1177,7 @@ func asConnectorStatus(record Record) ConnectorStatus {
 		Port:        record.AsString("port"),
 		Role:        record.AsString("role"),
 		Cost:        record.AsInt("cost"),
+		ConCount:    record.AsInt("data-connection-count"),
 		Status:      record.AsString("connectionStatus"),
 		Description: record.AsString("connectionMsg"),
 	}
@@ -1186,6 +1188,7 @@ func asConnector(record Record) Connector {
 		Name:           record.AsString("name"),
 		Host:           record.AsString("host"),
 		Port:           record.AsString("port"),
+		ConCount:       record.AsInt("data-connection-count"),
 		RouteContainer: record.AsBool("routeContainer"),
 		VerifyHostname: record.AsBool("verifyHostname"),
 		SslProfile:     record.AsString("sslProfile"),
@@ -1202,7 +1205,6 @@ func asSslProfile(record Record) SslProfile {
 }
 
 func asRecord(connector Connector) Record {
-
 	record := map[string]interface{}{}
 	record["name"] = connector.Name
 	record["role"] = string(connector.Role)
@@ -1220,6 +1222,9 @@ func asRecord(connector Connector) Record {
 	if connector.MaxSessionFrames > 0 {
 		record["maxSessionFrames"] = connector.MaxSessionFrames
 	}
+	if connector.ConCount > 0 {
+		record["dataConnectionCount"] = connector.ConCount
+	}
 
 	return record
 }

pkg/qdr/qdr.go

@@ -404,6 +404,7 @@ type Connector struct {
 	VerifyHostname   bool   `json:"verifyHostname,omitempty"`
 	SslProfile       string `json:"sslProfile,omitempty"`
 	LinkCapacity     int32  `json:"linkCapacity,omitempty"`
+	ConCount         int    `json:"dataConnectionCount,omitempty"`
 	MaxFrameSize     int    `json:"maxFrameSize,omitempty"`
 	MaxSessionFrames int    `json:"maxSessionFrames,omitempty"`
 }

pkg/utils/configs/config.go

@@ -25,6 +25,12 @@ func ConnectJson(host string) string {
 }
 
 func ConnectorConfig(connector *types.Connector) string {
+	// The data-connection-count expression will yield
+	// no output if .ConCount is 0.
+	// Almost no output. If it is on a line by itself,
+	// it will yield a blank line, which looks weird in the output.
+	// By putting it on the same line as 'cost' in this code,
+	// that blank output line is prevented.
 	config := `
 
 sslProfile {
@@ -39,7 +45,7 @@ connector {
     host: {{.Host}}
     port: {{.Port}}
     role: {{.Role}}
-    cost: {{.Cost}}
+    cost: {{.Cost}} {{if .ConCount}}{{printf "\n    dataConnectionCount:" }} {{.ConCount}}{{end}}
     sslProfile: {{.Name}}-profile
 }
 

test/integration/acceptance/custom/edge_connectivity/edge_connectivity.go

@@ -14,8 +14,6 @@ import (
 	"gotest.tools/assert"
 )
 
-var fp = fmt.Fprintf
-
 type TestCase struct {
 	name               string
 	diagram            []string