blob: 2294de6f6541f9b1db6a06503315f1320a802431 [file]
/*
* 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 cloudstack
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"strings"
"time"
"github.com/apache/cloudstack-go/v2/cloudstack"
"github.com/blang/semver/v4"
"gopkg.in/gcfg.v1"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/kubernetes"
cloudprovider "k8s.io/cloud-provider"
"k8s.io/klog/v2"
)
// ProviderName is the name of this cloud provider.
const ProviderName = "external-cloudstack"
// CSConfig wraps the config for the CloudStack cloud provider.
type CSConfig struct {
Global struct {
APIURL string `gcfg:"api-url"`
APIKey string `gcfg:"api-key"`
SecretKey string `gcfg:"secret-key"`
SSLNoVerify bool `gcfg:"ssl-no-verify"`
ProjectID string `gcfg:"project-id"`
Zone string `gcfg:"zone"`
Region string `gcfg:"region"`
}
}
// CSCloud is an implementation of Interface for CloudStack.
type CSCloud struct {
client *cloudstack.CloudStackClient
projectID string // If non-"", all resources will be created within this project
zone string
region string
version semver.Version
clientBuilder cloudprovider.ControllerClientBuilder
}
func init() {
cloudprovider.RegisterCloudProvider(ProviderName, func(config io.Reader) (cloudprovider.Interface, error) {
cfg, err := readConfig(config)
if err != nil {
return nil, err
}
return newCSCloud(cfg)
})
}
func readConfig(config io.Reader) (*CSConfig, error) {
cfg := &CSConfig{}
if config == nil {
return cfg, nil
}
if err := gcfg.ReadInto(cfg, config); err != nil {
return nil, fmt.Errorf("could not parse cloud provider config: %v", err)
}
return cfg, nil
}
// newCSCloud creates a new instance of CSCloud.
func newCSCloud(cfg *CSConfig) (*CSCloud, error) {
cs := &CSCloud{
projectID: cfg.Global.ProjectID,
zone: cfg.Global.Zone,
region: cfg.Global.Region,
version: semver.Version{},
}
if cfg.Global.APIURL != "" && cfg.Global.APIKey != "" && cfg.Global.SecretKey != "" {
cs.client = cloudstack.NewAsyncClient(cfg.Global.APIURL, cfg.Global.APIKey, cfg.Global.SecretKey, !cfg.Global.SSLNoVerify)
}
if cs.client == nil {
return nil, errors.New("no cloud provider config given")
}
version, err := cs.getManagementServerVersion()
if err != nil {
return nil, err
}
cs.version = version
return cs, nil
}
func (cs *CSCloud) getManagementServerVersion() (semver.Version, error) {
msServersResp, err := cs.client.Management.ListManagementServersMetrics(cs.client.Management.NewListManagementServersMetricsParams())
if err != nil {
return semver.Version{}, err
}
if msServersResp.Count == 0 {
return semver.Version{}, errors.New("no management servers found")
}
version := msServersResp.ManagementServersMetrics[0].Version
v, err := semver.ParseTolerant(strings.Join(strings.Split(version, ".")[0:3], "."))
if err != nil {
klog.Errorf("failed to parse management server version: %v", err)
return semver.Version{}, err
}
return v, nil
}
// Initialize passes a Kubernetes clientBuilder interface to the cloud provider
func (cs *CSCloud) Initialize(clientBuilder cloudprovider.ControllerClientBuilder, stop <-chan struct{}) {
cs.clientBuilder = clientBuilder
}
// LoadBalancer returns an implementation of LoadBalancer for CloudStack.
func (cs *CSCloud) LoadBalancer() (cloudprovider.LoadBalancer, bool) {
if cs.client == nil {
return nil, false
}
return cs, true
}
// Instances returns an implementation of Instances for CloudStack.
func (cs *CSCloud) Instances() (cloudprovider.Instances, bool) {
if cs.client == nil {
return nil, false
}
return cs, true
}
func (cs *CSCloud) InstancesV2() (cloudprovider.InstancesV2, bool) {
if cs.client == nil {
return nil, false
}
return cs, true
}
// Zones returns an implementation of Zones for CloudStack.
func (cs *CSCloud) Zones() (cloudprovider.Zones, bool) {
if cs.client == nil {
return nil, false
}
return cs, true
}
// Clusters returns an implementation of Clusters for CloudStack.
func (cs *CSCloud) Clusters() (cloudprovider.Clusters, bool) {
if cs.client == nil {
return nil, false
}
klog.Warning("This cloud provider doesn't support clusters")
return nil, false
}
// Routes returns an implementation of Routes for CloudStack.
func (cs *CSCloud) Routes() (cloudprovider.Routes, bool) {
if cs.client == nil {
return nil, false
}
klog.Warning("This cloud provider doesn't support routes")
return nil, false
}
// ProviderName returns the cloud provider ID.
func (cs *CSCloud) ProviderName() string {
return ProviderName
}
// HasClusterID returns true if the cluster has a clusterID
func (cs *CSCloud) HasClusterID() bool {
return true
}
// GetZone returns the Zone containing the region that the program is running in.
func (cs *CSCloud) GetZone(ctx context.Context) (cloudprovider.Zone, error) {
zone := cloudprovider.Zone{}
if cs.zone == "" {
// In Kubernetes pods, os.Hostname() returns the pod name, not the node hostname.
// We need to get the node name from the pod's spec.nodeName using the Kubernetes API.
nodeName, err := cs.getNodeNameFromPod(ctx)
if err != nil {
return zone, fmt.Errorf("failed to get node name for retrieving the zone: %v", err)
}
instance, count, err := cs.client.VirtualMachine.GetVirtualMachineByName(
nodeName,
cloudstack.WithProject(cs.projectID),
)
if err != nil {
if count == 0 {
return zone, fmt.Errorf("could not find CloudStack instance with name %s for retrieving the zone: %v", nodeName, err)
}
return zone, fmt.Errorf("error getting instance for retrieving the zone: %v", err)
}
cs.zone = instance.Zonename
}
klog.V(2).Infof("Current zone is %v", cs.zone)
zone.FailureDomain = cs.zone
zone.Region = cs.getRegionFromZone(cs.zone)
return zone, nil
}
// GetZoneByProviderID returns the Zone, found by using the provider ID.
func (cs *CSCloud) GetZoneByProviderID(ctx context.Context, providerID string) (cloudprovider.Zone, error) {
zone := cloudprovider.Zone{}
instance, count, err := cs.client.VirtualMachine.GetVirtualMachineByID(
cs.getInstanceIDFromProviderID(providerID),
cloudstack.WithProject(cs.projectID),
)
if err != nil {
if count == 0 {
return zone, fmt.Errorf("could not find node by ID: %v", providerID)
}
return zone, fmt.Errorf("error retrieving zone: %v", err)
}
klog.V(2).Infof("Current zone is %v", cs.zone)
zone.FailureDomain = instance.Zonename
zone.Region = cs.getRegionFromZone(instance.Zonename)
return zone, nil
}
// GetZoneByNodeName returns the Zone, found by using the node name.
func (cs *CSCloud) GetZoneByNodeName(ctx context.Context, nodeName types.NodeName) (cloudprovider.Zone, error) {
zone := cloudprovider.Zone{}
instance, count, err := cs.client.VirtualMachine.GetVirtualMachineByName(
string(nodeName),
cloudstack.WithProject(cs.projectID),
)
if err != nil {
if count == 0 {
return zone, fmt.Errorf("could not find node: %v", nodeName)
}
return zone, fmt.Errorf("error retrieving zone: %v", err)
}
klog.V(2).Infof("Current zone is %v", cs.zone)
zone.FailureDomain = instance.Zonename
zone.Region = cs.getRegionFromZone(instance.Zonename)
return zone, nil
}
// getNodeNameFromPod gets the node name where this pod is running by querying the Kubernetes API.
// It uses the pod's name and namespace (from environment variables or hostname) to look up the pod
// and retrieve its spec.nodeName field.
func (cs *CSCloud) getNodeNameFromPod(ctx context.Context) (string, error) {
if cs.clientBuilder == nil {
return "", fmt.Errorf("clientBuilder not initialized, cannot query Kubernetes API")
}
client, err := cs.clientBuilder.Client("cloud-controller-manager")
if err != nil {
return "", fmt.Errorf("failed to get Kubernetes client: %v", err)
}
// Get pod name and namespace
// In Kubernetes, the pod name is available as HOSTNAME environment variable
// or we can use os.Hostname() which returns the pod name
podName := os.Getenv("HOSTNAME")
if podName == "" {
var err error
podName, err = os.Hostname()
if err != nil {
return "", fmt.Errorf("failed to get pod name: %v", err)
}
}
// Get namespace from environment variable or default to kube-system for CCM
namespace := os.Getenv("POD_NAMESPACE")
if namespace == "" {
// Try reading from service account namespace file (available in pods)
if data, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace"); err == nil {
namespace = string(data)
} else {
// Default namespace for cloud controller manager
namespace = "kube-system"
}
}
// Get the pod object from Kubernetes API
pod, err := client.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{})
if err != nil {
return "", fmt.Errorf("failed to get pod %s/%s from Kubernetes API: %v", namespace, podName, err)
}
if pod.Spec.NodeName == "" {
return "", fmt.Errorf("pod %s/%s does not have a nodeName assigned yet", namespace, podName)
}
klog.V(4).Infof("found node name %s for pod %s/%s", pod.Spec.NodeName, namespace, podName)
return pod.Spec.NodeName, nil
}
// setServiceAnnotation updates a service annotation using the Kubernetes client.
// It uses a patch operation with retry logic to handle concurrent updates safely.
func (cs *CSCloud) setServiceAnnotation(ctx context.Context, service *corev1.Service, key, value string) error {
if cs.clientBuilder == nil {
klog.V(4).Infof("Client builder not available, skipping annotation update for service %s/%s", service.Namespace, service.Name)
return nil
}
client, err := cs.clientBuilder.Client("cloud-controller-manager")
if err != nil {
return fmt.Errorf("failed to get Kubernetes client: %v", err)
}
// First, check if the annotation already has the correct value to avoid unnecessary updates
svc, err := client.CoreV1().Services(service.Namespace).Get(ctx, service.Name, metav1.GetOptions{})
if err != nil {
if apierrors.IsNotFound(err) {
klog.V(4).Infof("Service %s/%s not found, skipping annotation update", service.Namespace, service.Name)
return nil
}
return fmt.Errorf("failed to get service: %v", err)
}
// Check if annotation already has the correct value
if svc.Annotations != nil {
if existingValue, exists := svc.Annotations[key]; exists && existingValue == value {
klog.V(4).Infof("Annotation %s already set to %s for service %s/%s", key, value, service.Namespace, service.Name)
return nil
}
}
// Use patch operation with retry logic to handle concurrent updates
return cs.patchServiceAnnotation(ctx, client, service.Namespace, service.Name, key, value)
}
// patchServiceAnnotation patches a service annotation using a JSON merge patch with retry logic.
// This method handles concurrent updates safely by retrying on conflicts.
func (cs *CSCloud) patchServiceAnnotation(ctx context.Context, client kubernetes.Interface, namespace, name, key, value string) error {
const maxRetries = 3
const retryDelay = 500 * time.Millisecond
// Prepare the patch payload - merge patch that updates only the specific annotation
// JSON merge patch will preserve other annotations while updating/adding this one
patchData := map[string]interface{}{
"metadata": map[string]interface{}{
"annotations": map[string]string{
key: value,
},
},
}
patchBytes, err := json.Marshal(patchData)
if err != nil {
return fmt.Errorf("failed to marshal patch data: %v", err)
}
for attempt := 0; attempt < maxRetries; attempt++ {
// Apply the patch using JSON merge patch type
// This is atomic and avoids race conditions by merging with existing annotations
_, err = client.CoreV1().Services(namespace).Patch(
ctx,
name,
types.MergePatchType,
patchBytes,
metav1.PatchOptions{},
)
if err == nil {
klog.V(4).Infof("Successfully set annotation %s=%s on service %s/%s", key, value, namespace, name)
return nil
}
// Handle conflict errors with retry logic
if apierrors.IsConflict(err) {
if attempt < maxRetries-1 {
klog.V(4).Infof("Conflict updating service %s/%s annotation, retrying (attempt %d/%d): %v", namespace, name, attempt+1, maxRetries, err)
time.Sleep(retryDelay)
continue
}
return fmt.Errorf("failed to update service annotation after %d retries due to conflicts: %v", maxRetries, err)
}
// Handle not found errors
if apierrors.IsNotFound(err) {
klog.V(4).Infof("Service %s/%s not found during patch, skipping annotation update", namespace, name)
return nil
}
// For other errors, return immediately
return fmt.Errorf("failed to patch service annotation: %v", err)
}
return fmt.Errorf("failed to update service annotation after %d attempts", maxRetries)
}
func (cs *CSCloud) getRegionFromZone(zone string) string {
if cs.region != "" {
return cs.region
}
return zone
}