admin-cli/client/meta.go

/*
 * 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 client

import (
	"context"
	"fmt"
	"reflect"
	"time"

	"github.com/apache/incubator-pegasus/admin-cli/util"
	"github.com/apache/incubator-pegasus/go-client/idl/admin"
	"github.com/apache/incubator-pegasus/go-client/idl/base"
	"github.com/apache/incubator-pegasus/go-client/idl/replication"
	"github.com/apache/incubator-pegasus/go-client/session"
)

type BalanceType int

const (
	BalanceMovePri BalanceType = iota
	BalanceCopyPri
	BalanceCopySec
)

func (t BalanceType) String() string {
	switch t {
	case BalanceMovePri:
		return "MovePri"
	case BalanceCopyPri:
		return "CopyPri"
	case BalanceCopySec:
		return "CopySec"
	default:
		panic(fmt.Sprintf("unexpected BalanceType: %d", t))
	}
}

// Meta is a helper over pegasus-go-client's primitive session.MetaManager.
// It aims to provide an easy-to-use API that eliminates some boilerplate code, like
// context creation, request/response creation, etc.
type Meta interface {
	Close() error

	// ListAvailableApps lists only available tables.
	ListAvailableApps() ([]*admin.AppInfo, error)

	ListApps(status admin.AppStatus) ([]*admin.AppInfo, error)

	QueryConfig(tableName string) (*replication.QueryCfgResponse, error)

	MetaControl(level admin.MetaFunctionLevel) (oldLevel admin.MetaFunctionLevel, err error)

	QueryClusterInfo() (map[string]string, error)

	UpdateAppEnvs(tableName string, envs map[string]string) error

	ClearAppEnvs(tableName string, clearPrefix string) error

	DelAppEnvs(tableName string, keys []string) error

	CreateApp(tableName string, envs map[string]string, partitionCount int) (int32, error)

	DropApp(tableName string, reserveSeconds int64) error

	ModifyDuplication(tableName string, dupid int, status admin.DuplicationStatus) error

	AddDuplication(tableName string, remoteCluster string, duplicateCheckpoint bool) (*admin.DuplicationAddResponse, error)

	QueryDuplication(tableName string) (*admin.DuplicationQueryResponse, error)

	ListNodes() ([]*admin.NodeInfo, error)

	RecallApp(originTableID int, newTableName string) (*admin.AppInfo, error)

	Balance(gpid *base.Gpid, opType BalanceType, from *util.PegasusNode, to *util.PegasusNode) error

	Propose(gpid *base.Gpid, action admin.ConfigType, target *util.PegasusNode, node *util.PegasusNode) error

	StartBackupApp(tableID int, providerType string, backupPath string) (*admin.StartBackupAppResponse, error)

	QueryBackupStatus(tableID int, backupID int64) (*admin.QueryBackupStatusResponse, error)

	RestoreApp(oldClusterName string, oldTableName string, oldTableID int, backupID int64, providerType string,
		newTableName string, restorePath string, skipBadPartition bool, policyName string) (*admin.CreateAppResponse, error)

	StartPartitionSplit(tableName string, newPartitionCount int) error

	QuerySplitStatus(tableName string) (*admin.QuerySplitResponse, error)

	PausePartitionSplit(tableName string, parentPidx int) error

	RestartPartitionSplit(tableName string, parentPidx int) error

	CancelPartitionSplit(tableName string, oldPartitionCount int) error

	StartBulkLoad(tableName string, clusterName string, providerType string, rootPath string) error

	QueryBulkLoad(tableName string) (*admin.QueryBulkLoadResponse, error)

	PauseBulkLoad(tableName string) error

	RestartBulkLoad(tableName string) error

	CancelBulkLoad(tableName string, forced bool) error

	ClearBulkLoad(tableName string) error

	StartManualCompaction(tableName string, targetLevel int, maxRunningCount int, bottommost bool) error

	QueryManualCompaction(tableName string) (*admin.QueryAppManualCompactResponse, error)
}

type rpcBasedMeta struct {
	meta *session.MetaManager
}

// NewRPCBasedMeta creates the connection to meta.
func NewRPCBasedMeta(metaAddrs []string) Meta {
	return &rpcBasedMeta{
		meta: session.NewMetaManager(metaAddrs, session.NewNodeSession),
	}
}

// Some responses have not only error-code but also a string-type "hint" that can tells the error details.
// This function wraps the "hint" into error.
func wrapHintIntoError(hint string, err error) error {
	if err != nil {
		if hint != "" {
			return fmt.Errorf("%s [hint: %s]", err, hint)
		}
	}
	return err
}

func (m *rpcBasedMeta) Close() error {
	return m.meta.Close()
}

// `callback` always accepts non-nil `resp`.
func (m *rpcBasedMeta) callMeta(methodName string, req interface{}, callback func(resp interface{})) error {
	ctx, cancel := context.WithTimeout(context.Background(), rpcTimeout)
	defer cancel()

	ret := reflect.ValueOf(m.meta).MethodByName(methodName).Call([]reflect.Value{
		reflect.ValueOf(ctx),
		reflect.ValueOf(req),
	})

	// the last returned value is always error
	ierr := ret[len(ret)-1].Interface()
	var err error
	if ierr != nil {
		err = ierr.(error)
	}

	if len(ret) == 1 {
		return err
	}

	// len(ret) == 2
	if !ret[0].IsNil() {
		callback(ret[0].Interface())
	}
	return err
}

func (m *rpcBasedMeta) ListAvailableApps() ([]*admin.AppInfo, error) {
	return m.ListApps(admin.AppStatus_AS_AVAILABLE)
}

func (m *rpcBasedMeta) ListApps(status admin.AppStatus) ([]*admin.AppInfo, error) {
	var result []*admin.AppInfo
	req := &admin.ListAppsRequest{Status: status}
	err := m.callMeta("ListApps", req, func(resp interface{}) {
		result = resp.(*admin.ListAppsResponse).Infos
	})
	return result, err
}

func (m *rpcBasedMeta) QueryConfig(tableName string) (*replication.QueryCfgResponse, error) {
	var result *replication.QueryCfgResponse
	err := m.callMeta("QueryConfig", tableName, func(resp interface{}) {
		result = resp.(*replication.QueryCfgResponse)
	})
	if err == nil {
		if result.GetErr().Errno == base.ERR_OBJECT_NOT_FOUND.String() {
			return nil, fmt.Errorf("table(%s) doesn't exist", tableName)
		}
		if result.GetErr().Errno != base.ERR_OK.String() {
			return nil, fmt.Errorf("query config failed: %s", result.GetErr())
		}
	}
	return result, err
}

func (m *rpcBasedMeta) MetaControl(level admin.MetaFunctionLevel) (oldLevel admin.MetaFunctionLevel, err error) {
	req := &admin.MetaControlRequest{Level: level}
	err = m.callMeta("MetaControl", req, func(resp interface{}) {
		oldLevel = resp.(*admin.MetaControlResponse).OldLevel
	})
	return oldLevel, err
}

func (m *rpcBasedMeta) QueryClusterInfo() (map[string]string, error) {
	result := make(map[string]string)
	req := &admin.ClusterInfoRequest{}
	err := m.callMeta("QueryClusterInfo", req, func(resp interface{}) {
		keys := resp.(*admin.ClusterInfoResponse).Keys
		values := resp.(*admin.ClusterInfoResponse).Values
		for i := range keys {
			result[keys[i]] = values[i]
		}
	})
	return result, err
}

func (m *rpcBasedMeta) updateAppEnvs(req *admin.UpdateAppEnvRequest) error {
	var hint string
	err := m.callMeta("UpdateAppEnv", req, func(resp interface{}) {
		hint = resp.(*admin.UpdateAppEnvResponse).HintMessage
	})
	return wrapHintIntoError(hint, err)
}

func (m *rpcBasedMeta) UpdateAppEnvs(tableName string, envs map[string]string) error {
	var keys []string
	var values []string
	for key, value := range envs {
		keys = append(keys, key)
		values = append(values, value)
	}
	req := &admin.UpdateAppEnvRequest{
		AppName: tableName,
		Op:      admin.AppEnvOperation_APP_ENV_OP_SET,
		Keys:    keys,
		Values:  values,
	}
	return m.updateAppEnvs(req)
}

func (m *rpcBasedMeta) ClearAppEnvs(tableName string, clearPrefix string) error {
	req := &admin.UpdateAppEnvRequest{
		AppName:     tableName,
		Op:          admin.AppEnvOperation_APP_ENV_OP_CLEAR,
		ClearPrefix: &clearPrefix,
	}
	return m.updateAppEnvs(req)
}

func (m *rpcBasedMeta) DelAppEnvs(tableName string, keys []string) error {
	req := &admin.UpdateAppEnvRequest{
		AppName: tableName,
		Op:      admin.AppEnvOperation_APP_ENV_OP_DEL,
		Keys:    keys,
	}
	return m.updateAppEnvs(req)
}

func (m *rpcBasedMeta) CreateApp(tableName string, envs map[string]string, partitionCount int) (int32, error) {
	var appID int32
	req := &admin.CreateAppRequest{
		AppName: tableName,
		Options: &admin.CreateAppOptions{
			PartitionCount: int32(partitionCount),
			Envs:           envs,

			// constants
			ReplicaCount: 3,
			IsStateful:   true,
			AppType:      "pegasus",
		},
	}
	err := m.callMeta("CreateApp", req, func(resp interface{}) {
		appID = resp.(*admin.CreateAppResponse).Appid
	})
	return appID, err
}

func (m *rpcBasedMeta) DropApp(tableName string, reserveSeconds int64) error {
	req := &admin.DropAppRequest{
		AppName: tableName,
		Options: &admin.DropAppOptions{
			SuccessIfNotExist: true,
			ReserveSeconds:    &reserveSeconds,
		},
	}
	err := m.callMeta("DropApp", req, func(resp interface{}) {})
	return err
}

func (m *rpcBasedMeta) ModifyDuplication(tableName string, dupid int, status admin.DuplicationStatus) error {
	req := &admin.DuplicationModifyRequest{
		AppName: tableName,
		Dupid:   int32(dupid),
		Status:  &status,
	}
	err := m.callMeta("ModifyDuplication", req, func(resp interface{}) {})
	return err
}

func (m *rpcBasedMeta) AddDuplication(tableName string, remoteCluster string, duplicateCheckpoint bool) (*admin.DuplicationAddResponse, error) {
	var result *admin.DuplicationAddResponse
	req := &admin.DuplicationAddRequest{
		AppName:                 tableName,
		RemoteClusterName:       remoteCluster,
		IsDuplicatingCheckpoint: duplicateCheckpoint,
	}
	err := m.callMeta("AddDuplication", req, func(resp interface{}) {
		result = resp.(*admin.DuplicationAddResponse)
	})
	if result != nil && result.IsSetHint() {
		return result, wrapHintIntoError(*result.Hint, err)
	}
	return result, err
}

func (m *rpcBasedMeta) QueryDuplication(tableName string) (*admin.DuplicationQueryResponse, error) {
	var result *admin.DuplicationQueryResponse
	req := &admin.DuplicationQueryRequest{
		AppName: tableName,
	}
	err := m.callMeta("QueryDuplication", req, func(resp interface{}) {
		result = resp.(*admin.DuplicationQueryResponse)
	})
	return result, err
}

func (m *rpcBasedMeta) ListNodes() ([]*admin.NodeInfo, error) {
	var result []*admin.NodeInfo
	req := &admin.ListNodesRequest{
		Status: admin.NodeStatus_NS_INVALID,
	}
	err := m.callMeta("ListNodes", req, func(resp interface{}) {
		result = resp.(*admin.ListNodesResponse).Infos
	})
	return result, err
}

func (m *rpcBasedMeta) RecallApp(originTableID int, newTableName string) (*admin.AppInfo, error) {
	var result *admin.AppInfo
	req := &admin.RecallAppRequest{
		AppID:       int32(originTableID),
		NewAppName_: newTableName,
	}
	err := m.callMeta("RecallApp", req, func(resp interface{}) {
		result = resp.(*admin.RecallAppResponse).Info
	})
	return result, err
}

func getNodeAddress(n *util.PegasusNode) *base.RPCAddress {
	if n == nil {
		return &base.RPCAddress{}
	}
	return base.NewRPCAddress(n.IP, n.Port)
}

func newProposalAction(target *util.PegasusNode, node *util.PegasusNode, cfgType admin.ConfigType) *admin.ConfigurationProposalAction {
	return &admin.ConfigurationProposalAction{
		Target: getNodeAddress(target),
		Node:   getNodeAddress(node),
		Type:   cfgType,
	}
}

func (m *rpcBasedMeta) Balance(gpid *base.Gpid, opType BalanceType, from *util.PegasusNode, to *util.PegasusNode) error {
	req := &admin.BalanceRequest{
		Gpid: gpid,
	}

	var actions []*admin.ConfigurationProposalAction
	switch opType {
	case BalanceMovePri:
		actions = append(actions, newProposalAction(from, from, admin.ConfigType_CT_DOWNGRADE_TO_SECONDARY))
		actions = append(actions, newProposalAction(to, to, admin.ConfigType_CT_UPGRADE_TO_PRIMARY))
	case BalanceCopyPri:
		actions = append(actions, newProposalAction(from, to, admin.ConfigType_CT_ADD_SECONDARY_FOR_LB))
		actions = append(actions, newProposalAction(from, from, admin.ConfigType_CT_DOWNGRADE_TO_SECONDARY))
		actions = append(actions, newProposalAction(to, to, admin.ConfigType_CT_UPGRADE_TO_PRIMARY))
		actions = append(actions, newProposalAction(to, from, admin.ConfigType_CT_REMOVE))
	case BalanceCopySec:
		actions = append(actions, newProposalAction(nil, to, admin.ConfigType_CT_ADD_SECONDARY_FOR_LB))
		actions = append(actions, newProposalAction(nil, from, admin.ConfigType_CT_DOWNGRADE_TO_INACTIVE))
	default:
		return fmt.Errorf("illegal balance type %d", opType)
	}
	req.ActionList = actions

	err := m.callMeta("Balance", req, func(resp interface{}) {})
	return err
}

func (m *rpcBasedMeta) Propose(gpid *base.Gpid, action admin.ConfigType, target *util.PegasusNode, node *util.PegasusNode) error {
	req := &admin.BalanceRequest{
		Gpid: gpid,
		ActionList: []*admin.ConfigurationProposalAction{
			newProposalAction(target, node, action),
		},
	}
	err := m.callMeta("Balance", req, func(resp interface{}) {})
	return err
}

func (m *rpcBasedMeta) StartBackupApp(tableID int, providerType string, backupPath string) (*admin.StartBackupAppResponse, error) {
	req := &admin.StartBackupAppRequest{
		BackupProviderType: providerType,
		AppID:              int32(tableID),
	}
	if backupPath != "" {
		req.BackupPath = &backupPath
	}
	var result *admin.StartBackupAppResponse
	err := m.callMeta("StartBackupApp", req, func(resp interface{}) {
		result = resp.(*admin.StartBackupAppResponse)
	})
	return result, wrapHintIntoError(result.HintMessage, err)
}

func (m *rpcBasedMeta) QueryBackupStatus(tableID int, backupID int64) (*admin.QueryBackupStatusResponse, error) {
	var realBackupID *int64
	if backupID == 0 {
		realBackupID = nil
	} else {
		realBackupID = &backupID
	}
	req := &admin.QueryBackupStatusRequest{
		AppID:    int32(tableID),
		BackupID: realBackupID,
	}
	var result *admin.QueryBackupStatusResponse
	err := m.callMeta("QueryBackupStatus", req, func(resp interface{}) {
		result = resp.(*admin.QueryBackupStatusResponse)
	})
	return result, wrapHintIntoError(result.HintMessage, err)
}

func (m *rpcBasedMeta) RestoreApp(oldClusterName string, oldTableName string, oldTableID int,
	backupID int64, providerType string, newTableName string, restorePath string,
	skipBadPartition bool, policyName string) (*admin.CreateAppResponse, error) {
	req := &admin.RestoreAppRequest{
		ClusterName:        oldClusterName,
		PolicyName:         policyName,
		TimeStamp:          backupID,
		AppName:            oldTableName,
		AppID:              int32(oldTableID),
		NewAppName_:        newTableName,
		BackupProviderName: providerType,
		SkipBadPartition:   skipBadPartition,
		RestorePath:        &restorePath,
	}
	var result *admin.CreateAppResponse
	SetRPCTimeout(time.Second * 20)
	err := m.callMeta("RestoreApp", req, func(resp interface{}) {
		result = resp.(*admin.CreateAppResponse)
	})
	return result, err
}

func (m *rpcBasedMeta) StartPartitionSplit(tableName string, newPartitionCount int) error {
	req := &admin.StartPartitionSplitRequest{
		AppName:            tableName,
		NewPartitionCount_: int32(newPartitionCount),
	}
	var result *admin.StartPartitionSplitResponse
	err := m.callMeta("StartPartitionSplit", req, func(resp interface{}) {
		result = resp.(*admin.StartPartitionSplitResponse)
	})
	return wrapHintIntoError(result.GetHintMsg(), err)
}

func (m *rpcBasedMeta) QuerySplitStatus(tableName string) (*admin.QuerySplitResponse, error) {
	req := &admin.QuerySplitRequest{
		AppName: tableName,
	}
	var result *admin.QuerySplitResponse
	err := m.callMeta("QuerySplitStatus", req, func(resp interface{}) {
		result = resp.(*admin.QuerySplitResponse)
	})
	return result, wrapHintIntoError(result.GetHintMsg(), err)
}

func (m *rpcBasedMeta) PausePartitionSplit(tableName string, parentPidx int) error {
	req := &admin.ControlSplitRequest{
		AppName:     tableName,
		ControlType: admin.SplitControlType_PAUSE,
		ParentPidx:  int32(parentPidx),
	}
	var result *admin.ControlSplitResponse
	err := m.callMeta("ControlPartitionSplit", req, func(resp interface{}) {
		result = resp.(*admin.ControlSplitResponse)
	})
	return wrapHintIntoError(result.GetHintMsg(), err)
}

func (m *rpcBasedMeta) RestartPartitionSplit(tableName string, parentPidx int) error {
	req := &admin.ControlSplitRequest{
		AppName:     tableName,
		ControlType: admin.SplitControlType_RESTART,
		ParentPidx:  int32(parentPidx),
	}
	var result *admin.ControlSplitResponse
	err := m.callMeta("ControlPartitionSplit", req, func(resp interface{}) {
		result = resp.(*admin.ControlSplitResponse)
	})
	return wrapHintIntoError(result.GetHintMsg(), err)
}

func (m *rpcBasedMeta) CancelPartitionSplit(tableName string, oldPartitionCount int) error {
	var partitionCount int32 = int32(oldPartitionCount)
	req := &admin.ControlSplitRequest{
		AppName:           tableName,
		ControlType:       admin.SplitControlType_CANCEL,
		ParentPidx:        -1,
		OldPartitionCount: &partitionCount,
	}
	var result *admin.ControlSplitResponse
	err := m.callMeta("ControlPartitionSplit", req, func(resp interface{}) {
		result = resp.(*admin.ControlSplitResponse)
	})
	return wrapHintIntoError(result.GetHintMsg(), err)
}

func (m *rpcBasedMeta) StartBulkLoad(tableName string, clusterName string, providerType string, rootPath string) error {
	req := &admin.StartBulkLoadRequest{
		AppName:          tableName,
		ClusterName:      clusterName,
		FileProviderType: providerType,
		RemoteRootPath:   rootPath,
	}
	var result *admin.StartBulkLoadResponse
	err := m.callMeta("StartBulkLoad", req, func(resp interface{}) {
		result = resp.(*admin.StartBulkLoadResponse)
	})
	return wrapHintIntoError(result.GetHintMsg(), err)
}

func (m *rpcBasedMeta) QueryBulkLoad(tableName string) (*admin.QueryBulkLoadResponse, error) {
	req := &admin.QueryBulkLoadRequest{
		AppName: tableName,
	}
	var result *admin.QueryBulkLoadResponse
	err := m.callMeta("QueryBulkLoadStatus", req, func(resp interface{}) {
		result = resp.(*admin.QueryBulkLoadResponse)
	})
	return result, wrapHintIntoError(result.GetHintMsg(), err)
}

func (m *rpcBasedMeta) PauseBulkLoad(tableName string) error {
	req := &admin.ControlBulkLoadRequest{
		AppName: tableName,
		Type:    admin.BulkLoadControlType_BLC_PAUSE,
	}
	var result *admin.ControlBulkLoadResponse
	err := m.callMeta("ControlBulkLoad", req, func(resp interface{}) {
		result = resp.(*admin.ControlBulkLoadResponse)
	})
	return wrapHintIntoError(result.GetHintMsg(), err)
}

func (m *rpcBasedMeta) RestartBulkLoad(tableName string) error {
	req := &admin.ControlBulkLoadRequest{
		AppName: tableName,
		Type:    admin.BulkLoadControlType_BLC_RESTART,
	}
	var result *admin.ControlBulkLoadResponse
	err := m.callMeta("ControlBulkLoad", req, func(resp interface{}) {
		result = resp.(*admin.ControlBulkLoadResponse)
	})
	return wrapHintIntoError(result.GetHintMsg(), err)
}

func (m *rpcBasedMeta) CancelBulkLoad(tableName string, forced bool) error {
	cancelType := admin.BulkLoadControlType_BLC_CANCEL
	if forced {
		cancelType = admin.BulkLoadControlType_BLC_FORCE_CANCEL
	}
	req := &admin.ControlBulkLoadRequest{
		AppName: tableName,
		Type:    cancelType,
	}
	var result *admin.ControlBulkLoadResponse
	err := m.callMeta("ControlBulkLoad", req, func(resp interface{}) {
		result = resp.(*admin.ControlBulkLoadResponse)
	})
	return wrapHintIntoError(result.GetHintMsg(), err)
}

func (m *rpcBasedMeta) ClearBulkLoad(tableName string) error {
	req := &admin.ClearBulkLoadStateRequest{
		AppName: tableName,
	}
	var result *admin.ClearBulkLoadStateResponse
	err := m.callMeta("ClearBulkLoad", req, func(resp interface{}) {
		result = resp.(*admin.ClearBulkLoadStateResponse)
	})
	return wrapHintIntoError(result.GetHintMsg(), err)
}

func (m *rpcBasedMeta) StartManualCompaction(tableName string, targetLevel int, maxRunningCount int, bottommost bool) error {
	var level int32 = int32(targetLevel)
	var count int32 = int32(maxRunningCount)
	req := &admin.StartAppManualCompactRequest{
		AppName:         tableName,
		TargetLevel:     &level,
		Bottommost:      &bottommost,
		MaxRunningCount: &count,
	}
	var result *admin.StartAppManualCompactResponse
	err := m.callMeta("StartManualCompact", req, func(resp interface{}) {
		result = resp.(*admin.StartAppManualCompactResponse)
	})
	return wrapHintIntoError(result.GetHintMsg(), err)
}

func (m *rpcBasedMeta) QueryManualCompaction(tableName string) (*admin.QueryAppManualCompactResponse, error) {
	req := &admin.QueryAppManualCompactRequest{
		AppName: tableName,
	}
	var result *admin.QueryAppManualCompactResponse
	err := m.callMeta("QueryManualCompact", req, func(resp interface{}) {
		result = resp.(*admin.QueryAppManualCompactResponse)
	})
	return result, wrapHintIntoError(result.GetHintMsg(), err)
}