func convertLogLevel(logLevel string) string {
	switch strings.ToLower(logLevel) {
	case "debug":
		return "Debug"
	case "info":
		return "Info"
	case "warning":
		return "Warning"
	case "error":
		return "Error"
	case "fatal":
		return "Fatal"
	case "panic":
		return "Fatal"
	case "":
		return ""
	default:
		return "Info"
	}
}
convertLogLevel converts the v1 log level to the equivalent v3 log level. We
ignore errors, defaulting to info in the event of a conversion error.
package migrator
import (
	"context"
	"errors"
	"fmt"
	"math/rand"
	"strings"
	"time"
	"github.com/coreos/go-semver/semver"
	log "github.com/sirupsen/logrus"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/uuid"
	apiv3 "github.com/projectcalico/api/pkg/apis/projectcalico/v3"
	"github.com/projectcalico/api/pkg/lib/numorstring"
	libapiv3 "github.com/projectcalico/libcalico-go/lib/apis/v3"
	bapi "github.com/projectcalico/libcalico-go/lib/backend/api"
	"github.com/projectcalico/libcalico-go/lib/backend/model"
	"github.com/projectcalico/libcalico-go/lib/clientv3"
	cerrors "github.com/projectcalico/libcalico-go/lib/errors"
	"github.com/projectcalico/libcalico-go/lib/options"
	"github.com/projectcalico/libcalico-go/lib/upgrade/converters"
	"github.com/projectcalico/libcalico-go/lib/upgrade/migrator/clients"
	validatorv3 "github.com/projectcalico/libcalico-go/lib/validator/v3"
)
const (
	forceEnableReadyRetries = 30
	maxApplyRetries         = 5
	numAppliesPerUpdate     = 100
	retryInterval           = 5 * time.Second
	minUpgradeVersion = "2.6.5"
)
type Interface interface {
	ValidateConversion() (*MigrationData, error)
	IsDestinationEmpty() (bool, error)
	ShouldMigrate() (bool, error)
	CanMigrate() error
	Migrate() (*MigrationData, error)
	IsMigrationInProgress() (bool, error)
	Abort() error
	Complete() error
}
type StatusWriterInterface interface {
	Msg(string)
	Bullet(string)
	Error(string)
}
func New(clientv3 clientv3.Interface, clientv1 clients.V1ClientInterface, statusWriter StatusWriterInterface) Interface {
	return &migrationHelper{
		clientv3:     clientv3,
		clientv1:     clientv1,
		statusWriter: statusWriter,
	}
}
type migrationHelper struct {
	clientv3     clientv3.Interface
	clientv1     clients.V1ClientInterface
	statusWriter StatusWriterInterface
}
type ErrorType int
const (
	ErrorGeneric ErrorType = iota
	ErrorConvertingData
	ErrorMigratingData
)
type MigrationError struct {
	Err        error
	Type       ErrorType
	NeedsAbort bool
}
func (m MigrationError) Error() string {
	return m.Err.Error()
}
type MigrationData struct {
	Resources []converters.Resource
	NameConversions []NameConversion
	ConversionErrors []ConversionError
	ConvertedResourceValidationErrors []ConversionError
	NameClashes []NameClash
	HandledByPolicyCtrl []model.Key
}
func (c *MigrationData) HasErrors() bool {
	return len(c.ConversionErrors) != 0 ||
		len(c.ConvertedResourceValidationErrors) != 0 ||
		len(c.NameClashes) != 0
}
type ConversionError struct {
	Cause   error
	KeyV1   model.Key
	ValueV1 interface{}
	KeyV3   model.Key
	ValueV3 converters.Resource
}
type NameConversion struct {
	KeyV1 model.Key
	KeyV3 model.Key
}
type NameClash struct {
	KeyV1      model.Key
	KeyV3      model.Key
	OtherKeyV1 model.Key
}
func (m *migrationHelper) ValidateConversion() (*MigrationData, error) {
	m.status("Validating conversion of v1 data to v3")
	data, err := m.queryAndConvertResources()
	if err != nil {
		m.statusError("Unable to perform validation, please resolve errors and retry")
		m.statusBullet("Cause: %v", err)
		return nil, MigrationError{
			Type: ErrorGeneric,
			Err:  err,
		}
	}
	if data.HasErrors() {
		m.statusError("Error converting data, check output for details and resolve issues before starting upgrade")
		return data, MigrationError{
			Type: ErrorConvertingData,
			Err:  fmt.Errorf("error converting data: %v", err),
		}
	}
	m.statusBullet("data conversion successful")
	if !m.clientv1.IsKDD() && len(data.Resources) == 0 {
		m.statusError("No v1 resources detected: is the API configuration correctly configured?")
		return nil, MigrationError{
			Type: ErrorGeneric,
			Err:  errors.New("no v1 resources detected: is the API configuration correctly configured?"),
		}
	}
	m.status("Data conversion validated successfully")
	return data, nil
}
func (m *migrationHelper) IsDestinationEmpty() (bool, error) {
	m.status("Validating the v3 datastore")
	clean, err := m.v3DatastoreIsClean()
	if err != nil {
		m.statusError("Unable to validate the v3 datastore")
		m.statusBullet("Cause: %v", err)
		return false, MigrationError{
			Type: ErrorGeneric,
			Err:  fmt.Errorf("unable to validate the v3 datastore: %v", err),
		}
	}
	if clean {
		m.statusBullet("the v3 datastore is empty")
	} else {
		m.statusBullet("the v3 datastore is not empty")
	}
	return clean, nil
}
func (m *migrationHelper) Migrate() (*MigrationData, error) {
	if !m.clientv1.IsKDD() {
		m.status("Pausing Calico networking")
		if err := m.clearReadyV1(); err != nil {
			m.statusError("Unable to pause calico networking")
			return nil, MigrationError{
				Type: ErrorGeneric,
				Err:  fmt.Errorf("unable to pause calico networking: %v", err),
			}
		}
		m.status("Calico networking is now paused - waiting for 15s")
		time.Sleep(15 * time.Second)
	}
	m.status("Querying current v1 snapshot and converting to v3")
	data, err := m.queryAndConvertResources()
	if err != nil {
		m.statusError("Unable to convert the v1 snapshot to v3")
		m.statusBullet("cause: %v", err)
		return nil, m.abortAfterError(
			fmt.Errorf("error converting data: %v", err), ErrorGeneric,
		)
	}
	if data.HasErrors() {
		m.statusError("Error converting data - will attempt to abort upgrade")
		return nil, m.abortAfterError(
			fmt.Errorf("error converting data: %v", err), ErrorConvertingData,
		)
	}
	m.statusBullet("data converted successfully")
	m.status("Storing v3 data")
	if err = m.storeV3Resources(data); err != nil {
		m.statusError("Unable to store the v3 resources")
		m.statusBullet("cause: %v", err)
		return nil, m.abortAfterError(
			fmt.Errorf("error storing converted data: %v", err), ErrorMigratingData,
		)
	}
	m.status("Migrating IPAM data")
	if m.clientv1.IsKDD() {
		m.statusBullet("no data to migrate - not supported")
	} else if err = m.migrateIPAMData(); err != nil {
		m.statusError("Unable to migrate the v3 IPAM data")
		m.statusBullet("cause: %v", err)
		return nil, m.abortAfterError(
			fmt.Errorf("error migrating IPAM data: %v", err), ErrorMigratingData,
		)
	}
	m.status("Data migration from v1 to v3 successful")
	m.statusBullet("check the output for details of the migrated resources")
	m.statusBullet("continue by upgrading your calico/node versions to Calico v3.x")
	return data, nil
}
func (m *migrationHelper) abortAfterError(err error, errType ErrorType) error {
	if m.clientv1.IsKDD() {
		return MigrationError{Type: errType, Err: err}
	}
	if ae := m.Abort(); ae == nil {
		return MigrationError{Type: errType, Err: err}
	}
	return MigrationError{Type: errType, Err: err, NeedsAbort: true}
}
func (m *migrationHelper) IsMigrationInProgress() (bool, error) {
	migrateNeeded, err := m.ShouldMigrate()
	if err != nil {
		return false, fmt.Errorf("error checking migration progress status: %v", err)
	} else if migrateNeeded {
		ready, err := m.isReady()
		if err != nil {
			return false, fmt.Errorf("error checking migration progress status: %v", err)
		}
		return !ready, nil
	} else {
		return false, nil
	}
}
func (m *migrationHelper) Abort() error {
	m.status("Aborting upgrade")
	var err error
	if !m.clientv1.IsKDD() {
		m.status("Re-enabling Calico networking for v1")
		for i := 0; i < forceEnableReadyRetries; i++ {
			err = m.setReadyV1()
			if err == nil {
				break
			}
			time.Sleep(1 * time.Second)
		}
	}
	if err != nil {
		m.statusError("Failed to abort upgrade. Retry command.")
		m.statusBullet("cause: %v", err)
		return MigrationError{Type: ErrorGeneric, Err: err, NeedsAbort: true}
	}
	m.status("Upgrade aborted successfully")
	return nil
}
func (m *migrationHelper) Complete() error {
	m.status("Completing upgrade")
	var err error
	if !m.clientv1.IsKDD() {
		m.status("Enabling Calico networking for v3")
		for i := 0; i < forceEnableReadyRetries; i++ {
			err = m.setReadyV3(true)
			if err == nil {
				break
			}
			time.Sleep(1 * time.Second)
		}
	}
	if err != nil {
		m.statusError("Failed to complete upgrade. Retry command.")
		m.statusBullet("cause: %v", err)
		return MigrationError{Type: ErrorGeneric, Err: err}
	}
	m.status("Upgrade completed successfully")
	return nil
}
type policyCtrlFilterOut func(model.Key) bool
var noFilter = func(_ model.Key) bool { return false }
var filterGNP = func(k model.Key) bool {
	gk := k.(model.PolicyKey)
	return strings.HasPrefix(gk.Name, "knp.default.")
}
var filterProfile = func(k model.Key) bool {
	gk := k.(model.ProfileKey)
	return strings.HasPrefix(gk.Name, "k8s_ns.") || strings.HasPrefix(gk.Name, "openstack-sg-")
}
var filterWEP = func(k model.Key) bool {
	gk := k.(model.WorkloadEndpointKey)
	return gk.OrchestratorID == apiv3.OrchestratorOpenStack
}
type ic interface {
	IsClean() (bool, error)
}
func (m *migrationHelper) v3DatastoreIsClean() (bool, error) {
	bc := m.clientv3.(backendClientAccessor).Backend()
	if i, ok := bc.(ic); ok {
		return i.IsClean()
	}
	return true, nil
}
func (m *migrationHelper) queryAndConvertResources() (*MigrationData, error) {
	data := &MigrationData{}
	if err := m.queryAndConvertFelixConfigV1ToV3(data); err != nil {
		return nil, err
	}
	m.statusBullet("handling BGPConfiguration (global) resource")
	if err := m.queryAndConvertGlobalBGPConfigV1ToV3(data); err != nil {
		return nil, err
	}
	if m.clientv1.IsKDD() {
		m.statusBullet("skipping Node resources - these do not need migrating")
	} else {
		m.statusBullet("handling Node resources")
		if err := m.queryAndConvertV1ToV3Nodes(data); err != nil {
			return nil, err
		}
	}
	if m.clientv1.IsKDD() {
		m.statusBullet("skipping BGPPeer (global) resources - these do not need migrating")
	} else {
		m.statusBullet("handling BGPPeer (global) resources")
		if err := m.queryAndConvertV1ToV3Resources(
			data, model.GlobalBGPPeerListOptions{}, converters.BGPPeer{}, noFilter,
		); err != nil {
			return nil, err
		}
	}
	m.statusBullet("handling BGPPeer (node) resources")
	if err := m.queryAndConvertV1ToV3Resources(
		data, model.NodeBGPPeerListOptions{}, converters.BGPPeer{}, noFilter,
	); err != nil {
		return nil, err
	}
	if m.clientv1.IsKDD() {
		m.statusBullet("skipping HostEndpoint resources - not supported")
	} else {
		m.statusBullet("handling HostEndpoint resources")
		if err := m.queryAndConvertV1ToV3Resources(
			data, model.HostEndpointListOptions{}, converters.HostEndpoint{}, noFilter,
		); err != nil {
			return nil, err
		}
	}
	if m.clientv1.IsKDD() {
		m.statusBullet("skipping IPPool resources - these do not need migrating")
	} else {
		m.statusBullet("handling IPPool resources")
		if err := m.queryAndConvertV1ToV3Resources(
			data, model.IPPoolListOptions{}, converters.IPPool{}, noFilter,
		); err != nil {
			return nil, err
		}
	}
	if m.clientv1.IsKDD() {
		m.statusBullet("skipping GlobalNetworkPolicy resources - these do not need migrating")
	} else {
		m.statusBullet("handling GlobalNetworkPolicy resources")
		if err := m.queryAndConvertV1ToV3Resources(
			data, model.PolicyListOptions{}, converters.Policy{}, filterGNP,
		); err != nil {
			return nil, err
		}
	}
	if m.clientv1.IsKDD() {
		m.statusBullet("skipping Profile resources - these do not need migrating")
	} else {
		m.statusBullet("handling Profile resources")
		if err := m.queryAndConvertV1ToV3Resources(
			data, model.ProfileListOptions{}, converters.Profile{}, filterProfile,
		); err != nil {
			return nil, err
		}
	}
	if m.clientv1.IsKDD() {
		m.statusBullet("skipping WorkloadEndpoint resources - these do not need migrating")
	} else {
		m.statusBullet("handling WorkloadEndpoint resources")
		if err := m.queryAndConvertV1ToV3Resources(
			data, model.WorkloadEndpointListOptions{}, converters.WorkloadEndpoint{}, filterWEP,
		); err != nil {
			return nil, err
		}
	}
	return data, nil
}
func (m *migrationHelper) queryAndConvertV1ToV3Resources(
	data *MigrationData,
	listInterface model.ListInterface,
	converter converters.Converter,
	filterOut policyCtrlFilterOut,
) error {
	kvps, err := m.clientv1.List(listInterface)
	if err != nil {
		switch err.(type) {
		case cerrors.ErrorResourceDoesNotExist, cerrors.ErrorOperationNotSupported:
			return nil
		default:
			return err
		}
	}
	convertedNames := make(map[string]model.Key, len(kvps))
	for _, kvp := range kvps {
		if filterOut(kvp.Key) {
			log.Infof("Filter out Policy Controller created resource: %s", kvp.Key)
			data.HandledByPolicyCtrl = append(data.HandledByPolicyCtrl, kvp.Key)
			continue
		}
		r, err := converter.BackendV1ToAPIV3(kvp)
		if err != nil {
			data.ConversionErrors = append(data.ConversionErrors, ConversionError{
				KeyV1:   kvp.Key,
				ValueV1: kvp.Value,
				Cause:   err,
			})
			continue
		}
		valid := true
		convertedName := r.GetObjectMeta().GetNamespace() + "/" + r.GetObjectMeta().GetName()
		if k, ok := convertedNames[convertedName]; ok {
			data.NameClashes = append(data.NameClashes, NameClash{
				KeyV1:      kvp.Key,
				KeyV3:      resourceToKey(r),
				OtherKeyV1: k,
			})
			valid = false
		}
		convertedNames[convertedName] = kvp.Key
		if err := validatorv3.Validate(r); err != nil {
			data.ConvertedResourceValidationErrors = append(data.ConvertedResourceValidationErrors, ConversionError{
				KeyV1:   kvp.Key,
				ValueV1: kvp.Value,
				KeyV3:   resourceToKey(r),
				ValueV3: r,
				Cause:   err,
			})
			valid = false
		}
		if valid {
			data.Resources = append(data.Resources, r)
			data.NameConversions = append(data.NameConversions, NameConversion{
				KeyV1: kvp.Key,
				KeyV3: resourceToKey(r),
			})
		}
	}
	return nil
}
func (m *migrationHelper) queryAndConvertGlobalBGPConfigV1ToV3(data *MigrationData) error {
	globalBGPConfig := apiv3.NewBGPConfiguration()
	globalBGPConfig.Name = "default"
	log.Info("Converting BGP config -> BGPConfiguration(default)")
	var setValue bool
	if kvp, err := m.clientv1.Get(model.GlobalBGPConfigKey{Name: "AsNumber"}); err != nil {
		if _, ok := err.(cerrors.ErrorResourceDoesNotExist); !ok {
			return err
		}
		log.Info("No global default ASNumber configured")
	} else if kvp.Value.(string) != "" {
		asNum, err := numorstring.ASNumberFromString(kvp.Value.(string))
		if err != nil {
			log.WithError(err).WithField("ASNumber", kvp.Value).Info("Invalid global default ASNumber")
			data.ConversionErrors = append(data.ConversionErrors, ConversionError{
				ValueV1: kvp.Value,
				Cause:   fmt.Errorf("default ASNumber is not valid: %s", kvp.Value.(string)),
			})
			return err
		}
		globalBGPConfig.Spec.ASNumber = &asNum
		setValue = true
	}
	if kvp, err := m.clientv1.Get(model.GlobalBGPConfigKey{Name: "LogLevel"}); err != nil {
		if _, ok := err.(cerrors.ErrorResourceDoesNotExist); !ok {
			return err
		}
		log.Info("No global BGP log level configured")
	} else if kvp.Value.(string) != "" {
		globalBGPConfig.Spec.LogSeverityScreen = convertLogLevel(kvp.Value.(string))
		setValue = true
	}
	if kvp, err := m.clientv1.Get(model.GlobalBGPConfigKey{Name: "NodeMeshEnabled"}); err != nil {
		if _, ok := err.(cerrors.ErrorResourceDoesNotExist); !ok {
			return err
		}
		log.Info("No global node to node mesh enabled setting configured")
	} else if kvp.Value.(string) != "" {
		nodeMeshEnabled := strings.ToLower(kvp.Value.(string)) == "true"
		globalBGPConfig.Spec.NodeToNodeMeshEnabled = &nodeMeshEnabled
		setValue = true
	}
	if setValue {
		data.Resources = append(data.Resources, globalBGPConfig)
	}
	return nil
}
func (m *migrationHelper) queryAndConvertV1ToV3Nodes(data *MigrationData) error {
	err := m.queryAndConvertV1ToV3Resources(
		data, model.NodeListOptions{}, converters.Node{}, noFilter,
	)
	if err != nil {
		return err
	}
	kvps, err := m.clientv1.List(model.HostConfigListOptions{Name: "IpInIpTunnelAddr"})
	if err != nil {
		switch err.(type) {
		case cerrors.ErrorResourceDoesNotExist, cerrors.ErrorOperationNotSupported:
			return nil
		default:
			return err
		}
	}
	addrs := map[string]string{}
	for _, kvp := range kvps {
		k := kvp.Key.(model.HostConfigKey)
		addrs[converters.ConvertNodeName(k.Hostname)] = kvp.Value.(string)
	}
	for _, r := range data.Resources {
		if nr, ok := r.(*libapiv3.Node); ok {
			addr := addrs[nr.Name]
			if addr == "" || nr.Spec.BGP == nil {
				continue
			}
			nr.Spec.BGP.IPv4IPIPTunnelAddr = addr
		}
	}
	return nil
}
func (m *migrationHelper) ShouldMigrate() (bool, error) {
	ci, err := m.clientv3.ClusterInformation().Get(context.Background(), "default", options.GetOptions{})
	if err == nil && ci.Spec.CalicoVersion == "" {
		log.Debug("ClusterInformation contained empty CalicoVersion - treating as if ClusterInformation is not present")
	} else if err == nil {
		if yes, err := versionRequiresMigration(ci.Spec.CalicoVersion); err != nil {
			log.Errorf("Unexpected CalicoVersion '%s' in ClusterInformation: %v", ci.Spec.CalicoVersion, err)
			return false, fmt.Errorf("unexpected CalicoVersion '%s' in ClusterInformation: %v", ci.Spec.CalicoVersion, err)
		} else if yes {
			log.Debugf("ClusterInformation contained CalicoVersion '%s' and indicates migration is needed", ci.Spec.CalicoVersion)
			return true, nil
		}
		log.Debugf("ClusterInformation contained CalicoVersion '%s' and indicates migration is not needed", ci.Spec.CalicoVersion)
		return false, nil
	} else if _, ok := err.(cerrors.ErrorResourceDoesNotExist); !ok {
		return false, fmt.Errorf("unable to query ClusterInformation to determine Calico version: %v", err)
	}
	v, err := m.getV1ClusterVersion()
	if err != nil {
		if _, ok := err.(cerrors.ErrorResourceDoesNotExist); ok {
			log.Debugf("CalicoVersion does exist in the v1 or v3 data, no migration needed")
			return false, nil
		}
		return false, fmt.Errorf("unable to query global Felix configuration to determine Calico version: %v", err)
	}
	if yes, err := versionRequiresMigration(v); err != nil {
		log.Errorf("Unable to migrate data from version '%s': %v", v, err)
		return false, fmt.Errorf("unable to migrate data from version '%s': %v", v, err)
	} else if !yes {
		log.Errorf("Unexpected version in the global Felix configuration, currently at %s", v)
		return false, fmt.Errorf("unexpected Calico version '%s': migration to v3 should be from a tagged "+
			"release of Calico v%s+", v, minUpgradeVersion)
	}
	log.Debugf("GlobalConfig contained CalicoVersion '%s' and indicates migration is needed", v)
	return true, nil
}
func (m *migrationHelper) CanMigrate() error {
	m.status("Checking Calico version is suitable for migration")
	v, err := m.getV1ClusterVersion()
	if err != nil {
		if _, ok := err.(cerrors.ErrorResourceDoesNotExist); ok {
			m.statusBullet("no version found in the v1 API datastore")
			return errors.New("unable to determine the Calico API version: no version information found - this may be " +
				"due to a misconfiguration of the v1 API datastore access details")
		}
		m.statusBullet("unable to query Calico version in the v1 API datastore: %v", err)
		return err
	}
	m.statusBullet("determined Calico version of: %s", v)
	if yes, err := versionRequiresMigration(v); err != nil {
		m.statusBullet("unable to parse version")
		return fmt.Errorf("unable to migrate data from version '%s': %v", v, err)
	} else if !yes {
		m.statusBullet("unexpected version in the v1 API datastore: should not have a version v3.0+")
		return fmt.Errorf("unexpected Calico version '%s': migration to v3 should be from a tagged "+
			"release of Calico v%s+", v, minUpgradeVersion)
	}
	m.statusBullet("the v1 API data can be migrated to the v3 API")
	return nil
}
func (m *migrationHelper) getV1ClusterVersion() (string, error) {
	kv, err := m.clientv1.Get(model.GlobalConfigKey{Name: "CalicoVersion"})
	if err == nil {
		return kv.Value.(string), nil
	}
	return "", err
}
func versionRequiresMigration(v string) (bool, error) {
	sv, err := semver.NewVersion(strings.TrimPrefix(v, "v"))
	if err != nil {
		log.Warnf("unable to parse Calico version %s: %v", v, err)
		return false, errors.New("unable to parse the version")
	}
	if sv.Major >= 3 {
		log.Debugf("major version is already >= 3: %s", v)
		return false, nil
	}
	sv.PreRelease = semver.PreRelease("")
	svMin := semver.New(minUpgradeVersion)
	if sv.Compare(*svMin) >= 0 {
		return true, nil
	}
	log.Infof("cannot migrate from version %s: migration to v3 requires a tagged release of "+
		"Calico v%s+", v, minUpgradeVersion)
	return false, fmt.Errorf("migration to v3 requires a tagged release of Calico v%s+", minUpgradeVersion)
}