added loadbalancer

pkg/pvecloud/loadbalancer.go

@@ -0,0 +1,119 @@
+/*
+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.
+*/
+
+package pvecloud
+
+import (
+	"context"
+
+	apiclient "git.giftfish.de/ston1th/haproxy-lb/pkg/api/client"
+	"git.giftfish.de/ston1th/haproxy-lb/pkg/api/v1/client"
+	"git.giftfish.de/ston1th/haproxy-lb/pkg/api/v1/schema"
+	v1 "k8s.io/api/core/v1"
+)
+
+type loadbalancer struct {
+	client *client.Client
+}
+
+func newLoadbalancer(client *client.Client) *loadbalancer {
+	return &loadbalancer{client}
+}
+
+func name(service *v1.Service) string {
+	return service.Namespace + "_" + service.Name
+}
+
+func fullname(clusterName string, service *v1.Service) string {
+	return clusterName + "/" + name(service)
+}
+
+func lbStatus(ip string) *v1.LoadBalancerStatus {
+	return &v1.LoadBalancerStatus{
+		Ingress: []v1.LoadBalancerIngress{
+			{IP: ip},
+		},
+	}
+}
+
+func (lb *loadbalancer) GetLoadBalancer(ctx context.Context, clusterName string, service *v1.Service) (status *v1.LoadBalancerStatus, exists bool, err error) {
+	l, err := lb.client.GetLoadBalancer(ctx, clusterName, name(service))
+	if err != nil {
+		return
+	}
+	exists = true
+	status = lbStatus(l.IP)
+	return
+}
+
+func (lb *loadbalancer) GetLoadBalancerName(ctx context.Context, clusterName string, service *v1.Service) string {
+	return fullname(clusterName, service)
+}
+
+func (lb *loadbalancer) EnsureLoadBalancer(ctx context.Context, clusterName string, service *v1.Service, nodes []*v1.Node) (status *v1.LoadBalancerStatus, err error) {
+	// TODO service annotations
+	var slb schema.LoadBalancer
+	for _, port := range service.Spec.Ports {
+		sp := schema.Port{Port: int(port.Port)}
+		for _, node := range nodes {
+			if len(node.Status.Addresses) == 0 {
+				continue
+			}
+			ip := ""
+			for _, a := range node.Status.Addresses {
+				if a.Type == v1.NodeInternalIP {
+					ip = a.Address
+					break
+				} else if a.Type == v1.NodeHostName {
+					ip = a.Address
+					break
+				} else if a.Type == v1.NodeExternalIP {
+					ip = a.Address
+					break
+				}
+			}
+			if ip == "" {
+				continue
+			}
+			sp.Servers = append(sp.Servers, schema.Server{
+				Name: node.Name,
+				IP:   ip,
+				Port: int(port.NodePort),
+			})
+		}
+		slb.Ports = append(slb.Ports, sp)
+	}
+	n := name(service)
+	err = lb.client.CreateLoadBalancer(ctx, clusterName, n, slb)
+	if err != nil {
+		return
+	}
+	l, err := lb.client.GetLoadBalancer(ctx, clusterName, n)
+	if err != nil {
+		return
+	}
+	status = lbStatus(l.IP)
+	return
+}
+
+func (lb *loadbalancer) UpdateLoadBalancer(ctx context.Context, clusterName string, service *v1.Service, nodes []*v1.Node) error {
+	_, err := lb.EnsureLoadBalancer(ctx, clusterName, service, nodes)
+	return err
+}
+
+func (lb *loadbalancer) EnsureLoadBalancerDeleted(ctx context.Context, clusterName string, service *v1.Service) error {
+	err := lb.client.DeleteLoadBalancer(ctx, clusterName, name(service))
+	if err == apiclient.ErrNotFound {
+		return nil
+	}
+	return err
+}