func (s *ServerGRPC) validateInputParamsForQuery(
	req *lcmservice.QueryRequest) (hostIp string, appInsId string, err error) {
	accessToken := req.GetAccessToken()
	err = util.ValidateAccessToken(accessToken, []string{util.MecmTenantRole, util.MecmGuestRole, util.MecmAdminRole})
	if err != nil {
		return "", "", s.logError(status.Error(codes.InvalidArgument,
			util.AccssTokenIsInvalid))
	}
	hostIp = req.GetHostIp()
	err = util.ValidateIpv4Address(hostIp)
	if err != nil {
		return "", "", s.logError(status.Error(codes.InvalidArgument, util.HostIpIsInvalid))
	}
	appInsId = req.GetAppInstanceId()
	err = util.ValidateUUID(appInsId)
	if err != nil {
		return "", "", s.logError(status.Error(codes.InvalidArgument, util.AppInsIdValid))
	}
	return hostIp, appInsId, nil
}
Validate input parameters for Query
package server
import (
	"archive/zip"
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"golang.org/x/time/rate"
	"google.golang.org/grpc/peer"
	"google.golang.org/grpc/tap"
	"io"
	"io/ioutil"
	"k8splugin/conf"
	"k8splugin/internal/lcmservice"
	"k8splugin/models"
	"k8splugin/pgdb"
	"k8splugin/pkg/adapter"
	"k8splugin/util"
	"net"
	"os"
	"path"
	"path/filepath"
	"strings"
	log "github.com/sirupsen/logrus"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials"
	_ "google.golang.org/grpc/encoding/gzip"
	"google.golang.org/grpc/status"
)
var (
	KubeconfigPath = "/usr/app/artifacts/config/"
	appPackagesBasePath = "/usr/app/artifacts/packages/"
)
type ServerGRPC struct {
	server       *grpc.Server
	port         string
	address      string
	certificate  string
	key          string
	db           pgdb.Database
	serverConfig *conf.ServerConfigurations
}
type ServerGRPCConfig struct {
	Port         string
	Address      string
	ServerConfig *conf.ServerConfigurations
}
type RateLimit struct {
	lim *rate.Limiter
}
func NewRateLimit() *RateLimit {
	return &RateLimit{rate.NewLimiter(1, 200)}
}
func NewServerGRPC(cfg ServerGRPCConfig) (s ServerGRPC) {
	s.port = cfg.Port
	s.address = cfg.Address
	s.certificate = cfg.ServerConfig.CertFilePath
	s.key = cfg.ServerConfig.KeyFilePath
	s.serverConfig = cfg.ServerConfig
	dbAdapter, err := pgdb.GetDbAdapter(cfg.ServerConfig)
	if err != nil {
		log.Error("Failed to get database")
		os.Exit(1)
	}
	s.db = dbAdapter
	log.Infof("Binding is successful")
	return
}
func (s *ServerGRPC) Listen() (err error) {
	var (
		listener net.Listener
	)
	listener, err = net.Listen("tcp", s.address+":"+s.port)
	if err != nil {
		log.Error("failed to listen on specified port")
		return err
	}
	if !s.serverConfig.SslNotEnabled {
		tlsConfig, err := util.GetTLSConfig(s.serverConfig, s.certificate, s.key)
		if err != nil {
			log.Error("failed to load certificates")
			return err
		}
		creds := credentials.NewTLS(tlsConfig)
		s.server = grpc.NewServer(grpc.Creds(creds), grpc.InTapHandle(NewRateLimit().Handler))
	} else {
		s.server = grpc.NewServer(grpc.InTapHandle(NewRateLimit().Handler))
	}
	lcmservice.RegisterAppLCMServer(s.server, s)
	log.Infof("Server registered with GRPC")
	err = s.server.Serve(listener)
	if err != nil {
		log.Error("failed to listen for GRPC connections.")
		return err
	}
	log.Info("Server started listening on configured port")
	return
}
func (t *RateLimit) Handler(ctx context.Context, info *tap.Info) (context.Context, error) {
	if !t.lim.Allow() {
		return nil, status.Errorf(codes.ResourceExhausted, "service is over rate limit")
	}
	return ctx, nil
}
func (s *ServerGRPC) WorkloadEvents(ctx context.Context, req *lcmservice.WorkloadEventsRequest) (resp *lcmservice.WorkloadEventsResponse, err error) {
	resp = &lcmservice.WorkloadEventsResponse{
		Response: util.Failure,
	}
	err = s.displayReceivedMsg(ctx, util.WorkloadEvents)
	if err != nil {
		s.displayResponseMsg(ctx, util.WorkloadEvents, util.FailedToDispRecvMsg)
		return resp, err
	}
	hostIp, appInsId, err := s.validateInputParamsForPodDesc(req)
	if err != nil {
		s.displayResponseMsg(ctx, util.WorkloadEvents, util.FailedToValInputParams)
		return resp, err
	}
	client, err := adapter.GetClient(util.DeployType, hostIp)
	if err != nil {
		s.displayResponseMsg(ctx, util.WorkloadEvents, util.FailedToGetClient)
		return resp, err
	}
	appInstanceRecord := &models.AppInstanceInfo{
		AppInsId: appInsId,
	}
	readErr := s.db.ReadData(appInstanceRecord, util.AppInsId)
	if readErr != nil {
		log.Error(util.AppRecordDoesNotExit)
		s.displayResponseMsg(ctx, util.Query, util.AppRecordDoesNotExit)
		return resp, err
	}
	r, err := client.WorkloadEvents(appInstanceRecord.WorkloadId, appInstanceRecord.Namespace)
	if err != nil {
		s.displayResponseMsg(ctx, util.WorkloadEvents, "failed to get pod describe information")
		return resp, err
	}
	resp = &lcmservice.WorkloadEventsResponse{
		Response: r,
	}
	s.handleLoggingForSuccess(ctx, util.WorkloadEvents, "Pod description is successful")
	return resp, nil
}
func (s *ServerGRPC) Query(ctx context.Context, req *lcmservice.QueryRequest) (resp *lcmservice.QueryResponse, err error) {
	resp = &lcmservice.QueryResponse{
		Response: util.Failure,
	}
	err = s.displayReceivedMsg(ctx, util.Query)
	if err != nil {
		s.displayResponseMsg(ctx, util.Query, util.FailedToDispRecvMsg)
		return resp, err
	}
	hostIp, appInsId, err := s.validateInputParamsForQuery(req)
	if err != nil {
		s.displayResponseMsg(ctx, util.Query, util.FailedToValInputParams)
		return resp, err
	}
	client, err := adapter.GetClient(util.DeployType, hostIp)
	if err != nil {
		s.displayResponseMsg(ctx, util.Query, util.FailedToGetClient)
		return resp, err
	}
	appInstanceRecord := &models.AppInstanceInfo{
		AppInsId: appInsId,
	}
	readErr := s.db.ReadData(appInstanceRecord, util.AppInsId)
	if readErr != nil {
		log.Error(util.AppRecordDoesNotExit)
		s.displayResponseMsg(ctx, util.Query, util.AppRecordDoesNotExit)
		return resp, err
	}
	r, err := client.Query(appInstanceRecord.WorkloadId, appInstanceRecord.Namespace)
	if err != nil {
		log.Errorf("Chart not found for workloadId: %s. Err: %s", appInstanceRecord.WorkloadId, err)
		s.displayResponseMsg(ctx, util.Query, "chart not found for workloadId")
		return resp, err
	}
	resp = &lcmservice.QueryResponse{
		Response: r,
	}
	s.handleLoggingForSuccess(ctx, util.Query, "Query pod statistics is successful")
	return resp, nil
}
func (s *ServerGRPC) QueryKPI(ctx context.Context, req *lcmservice.QueryKPIRequest) (resp *lcmservice.QueryKPIResponse, err error) {
	resp = &lcmservice.QueryKPIResponse{
		Response: util.Failure,
	}
	err = s.displayReceivedMsg(ctx, util.QueryKPI)
	if err != nil {
		s.displayResponseMsg(ctx, util.QueryKPI, util.FailedToDispRecvMsg)
		return resp, err
	}
	hostIp, err := s.validateInputParamsForQueryKPI(req)
	if err != nil {
		s.displayResponseMsg(ctx, util.QueryKPI, util.FailedToValInputParams)
		return resp, err
	}
	client, err := adapter.GetClient(util.DeployType, hostIp)
	if err != nil {
		s.displayResponseMsg(ctx, util.QueryKPI, util.FailedToGetClient)
		return resp, err
	}
	r, err := client.QueryKPI()
	if err != nil {
		s.displayResponseMsg(ctx, util.QueryKPI, "Query kpi statistics is failed")
		return resp, err
	}
	resp = &lcmservice.QueryKPIResponse{
		Response: r,
	}
	s.handleLoggingForSuccess(ctx, util.QueryKPI, "Query kpi statistics is successful")
	return resp, nil
}
func (s *ServerGRPC) Terminate(ctx context.Context,
	req *lcmservice.TerminateRequest) (resp *lcmservice.TerminateResponse, err error) {
	resp = &lcmservice.TerminateResponse{
		Status: util.Failure,
	}
	err = s.displayReceivedMsg(ctx, util.Terminate)
	if err != nil {
		s.displayResponseMsg(ctx, util.Terminate, util.FailedToDispRecvMsg)
		return resp, err
	}
	hostIp, appInsId, err := s.validateInputParamsForTerm(req)
	if err != nil {
		s.displayResponseMsg(ctx, util.Terminate, util.FailedToValInputParams)
		return resp, err
	}
	appInstanceRecord := &models.AppInstanceInfo{
		AppInsId: appInsId,
	}
	readErr := s.db.ReadData(appInstanceRecord, util.AppInsId)
	if readErr != nil {
		log.Error(util.AppRecordDoesNotExit)
		s.displayResponseMsg(ctx, util.Terminate, util.AppRecordDoesNotExit)
		return resp, err
	}
	client, err := adapter.GetClient(util.DeployType, hostIp)
	if err != nil {
		s.displayResponseMsg(ctx, util.Terminate, util.FailedToGetClient)
		return resp, err
	}
	err = client.UnDeploy(appInstanceRecord.WorkloadId, appInstanceRecord.Namespace)
	if err != nil {
		log.Errorf("Chart not found for workloadId: %s. Err: %s", appInstanceRecord.WorkloadId, err)
		s.displayResponseMsg(ctx, util.Terminate, "chart not found for workloadId")
		return resp, err
	}
	err = s.deleteAppInfoRecord(appInsId)
	if err != nil {
		s.displayResponseMsg(ctx, util.Terminate, "failed to delete app info record from database")
		return resp, err
	}
	resp = &lcmservice.TerminateResponse{
		Status: util.Success,
	}
	s.handleLoggingForSuccess(ctx, util.Terminate, "Termination is successful")
	return resp, nil
}
func (s *ServerGRPC) Instantiate(ctx context.Context,
	req *lcmservice.InstantiateRequest) (resp *lcmservice.InstantiateResponse, err error) {
	resp = &lcmservice.InstantiateResponse{
		Status: util.Failure,
	}
	err = s.displayReceivedMsg(ctx, util.Instantiate)
	if err != nil {
		s.displayResponseMsg(ctx, util.Instantiate, util.FailedToDispRecvMsg)
		return resp, err
	}
	tenantId, packageId, hostIp, appInsId, ak, sk, err := s.validateInputParamsForInstantiate(req)
	if err != nil {
		s.displayResponseMsg(ctx, util.Instantiate, util.FailedToValInputParams)
		return resp, err
	}
	appPkgRecord := &models.AppPackage{
		AppPkgId: packageId + tenantId + hostIp,
	}
	readErr := s.db.ReadData(appPkgRecord, util.AppPkgId)
	if readErr != nil {
		log.Error(util.AppPkgRecordDoesNotExit)
		s.displayResponseMsg(ctx, util.Instantiate, util.AppPkgRecordDoesNotExit)
		return resp, err
	}
	client, err := adapter.GetClient(util.DeployType, hostIp)
	if err != nil {
		s.displayResponseMsg(ctx, util.Instantiate, util.FailedToGetClient)
		return resp, err
	}
	releaseName, namespace, err := client.Deploy(appPkgRecord, appInsId, ak, sk, s.db)
	if err != nil {
		log.Info("instantiation failed")
		s.displayResponseMsg(ctx, util.Instantiate, "instantiation failed")
		return resp, err
	}
	err = s.insertOrUpdateAppInsRecord(appInsId, hostIp, releaseName, namespace)
	if err != nil {
		s.displayResponseMsg(ctx, util.Instantiate, "failed to insert or update app record")
		return resp, err
	}
	log.Info("successful instantiation")
	resp.Status = util.Success
	s.handleLoggingForSuccess(ctx, util.Instantiate, "Application instantiated successfully")
	return resp, nil
}
func (s *ServerGRPC) UploadConfig(stream lcmservice.AppLCM_UploadConfigServer) (err error) {
	var res lcmservice.UploadCfgResponse
	res.Status = util.Failure
	ctx := stream.Context()
	err = s.displayReceivedMsg(ctx, util.UploadConfig)
	if err != nil {
		s.displayResponseMsg(ctx, util.UploadConfig, util.FailedToDispRecvMsg)
		sendUploadCfgResponse(stream, &res)
		return err
	}
	hostIp, err := s.validateInputParamsForUploadCfg(stream)
	if err != nil {
		s.displayResponseMsg(ctx, util.UploadConfig, util.FailedToValInputParams)
		sendUploadCfgResponse(stream, &res)
		return
	}
	file, err := s.getUploadConfigFile(stream)
	if err != nil {
		s.displayResponseMsg(ctx, util.UploadConfig, "failed to get upload config file")
		sendUploadCfgResponse(stream, &res)
		return err
	}
	if !util.CreateDir(KubeconfigPath) {
		s.displayResponseMsg(ctx, util.UploadConfig, "failed to create config directory")
		sendUploadCfgResponse(stream, &res)
		return err
	}
	configPath := KubeconfigPath + hostIp
	newFile, err := os.Create(configPath)
	if err != nil {
		s.displayResponseMsg(ctx, util.UploadConfig, "failed to create config path")
		sendUploadCfgResponse(stream, &res)
		return err
	}
	if len(file.Bytes()) > util.MaxConfigFile {
		s.displayResponseMsg(ctx, util.UploadConfig, "file size is larger than max size")
		sendUploadCfgResponse(stream, &res)
		return err
	}
	defer newFile.Close()
	_, err = newFile.Write(file.Bytes())
	if err != nil {
		s.displayResponseMsg(ctx, util.UploadConfig, "config IO operation error")
		sendUploadCfgResponse(stream, &res)
		return err
	}
	res.Status = util.Success
	sendUploadCfgResponse(stream, &res)
	s.handleLoggingForSuccess(ctx, util.UploadConfig, "Upload config is successful")
	return nil
}
func (s *ServerGRPC) RemoveConfig(ctx context.Context,
	request *lcmservice.RemoveCfgRequest) (*lcmservice.RemoveCfgResponse, error) {
	resp := &lcmservice.RemoveCfgResponse{
		Status: util.Failure,
	}
	err := s.displayReceivedMsg(ctx, util.RemoveConfig)
	if err != nil {
		s.displayResponseMsg(ctx, util.RemoveConfig, util.FailedToDispRecvMsg)
		return resp, err
	}
	hostIp, err := s.validateInputParamsForRemoveCfg(request)
	if err != nil {
		s.displayResponseMsg(ctx, util.RemoveConfig, util.FailedToValInputParams)
		return resp, err
	}
	configPath := KubeconfigPath + hostIp
	err = os.Remove(configPath)
	if err != nil {
		log.Error("failed to remove host config file")
		s.displayResponseMsg(ctx, util.RemoveConfig, "failed to remove host config file")
		return resp, err
	}
	resp = &lcmservice.RemoveCfgResponse{
		Status: util.Success,
	}
	s.handleLoggingForSuccess(ctx, util.UploadConfig, "Remove config is successful")
	return resp, nil
}
func (s *ServerGRPC) validateInputParamsForRemoveCfg(request *lcmservice.RemoveCfgRequest) (string, error) {
	accessToken := request.GetAccessToken()
	err := util.ValidateAccessToken(accessToken, []string{util.MecmTenantRole, util.MecmAdminRole})
	if err != nil {
		if err.Error() == util.Forbidden {
			return "", s.logError(status.Error(codes.PermissionDenied, util.Forbidden))
		} else {
			return "", s.logError(status.Error(codes.InvalidArgument, util.AccssTokenIsInvalid))
		}
	}
	hostIp := request.GetHostIp()
	err = util.ValidateIpv4Address(hostIp)
	if err != nil {
		return "", s.logError(status.Error(codes.InvalidArgument, util.HostIpIsInvalid))
	}
	return hostIp, nil
}
func (s *ServerGRPC) contextError(ctx context.Context) error {
	switch ctx.Err() {
	case context.Canceled:
		return s.logError(status.Error(codes.Canceled, "request is canceled"))
	case context.DeadlineExceeded:
		return s.logError(status.Error(codes.DeadlineExceeded, "deadline is exceeded"))
	default:
		return nil
	}
}
func (s *ServerGRPC) logError(err error) error {
	if err != nil {
		log.Errorf("Error Information: %v", err)
	}
	return err
}
func (s *ServerGRPC) validateInputParamsForTerm(
	req *lcmservice.TerminateRequest) (hostIp string, appInsId string, err error) {
	accessToken := req.GetAccessToken()
	err = util.ValidateAccessToken(accessToken, []string{util.MecmTenantRole, util.MecmAdminRole})
	if err != nil {
		if err.Error() == util.Forbidden {
			return "", "", s.logError(status.Error(codes.PermissionDenied, util.Forbidden))
		} else {
			return "", "", s.logError(status.Error(codes.InvalidArgument,
				util.AccssTokenIsInvalid))
		}
	}
	hostIp = req.GetHostIp()
	err = util.ValidateIpv4Address(hostIp)
	if err != nil {
		return "", "", s.logError(status.Error(codes.InvalidArgument,
			util.HostIpIsInvalid))
	}
	appInsId = req.GetAppInstanceId()
	err = util.ValidateUUID(appInsId)
	if err != nil {
		return "", "", s.logError(status.Error(codes.InvalidArgument,
			util.AppInsIdValid))
	}
	return hostIp, appInsId, nil
}
func (s *ServerGRPC) validateInputParamsForInstantiate(
	req *lcmservice.InstantiateRequest) (tenantId string, packageId string, hostIp string, appInsId string, ak string, sk string, err error) {
	accessToken := req.GetAccessToken()
	err = util.ValidateAccessToken(accessToken, []string{util.MecmTenantRole, util.MecmAdminRole})
	if err != nil {
		if err.Error() == util.Forbidden {
			return "", "", "", "",  "", "", s.logError(status.Error(codes.PermissionDenied, util.Forbidden))
		} else {
			return "", "", "", "",  "", "", s.logError(status.Error(codes.InvalidArgument,
				util.AccssTokenIsInvalid))
		}
	}
	hostIp = req.GetHostIp()
	err = util.ValidateIpv4Address(hostIp)
	if err != nil {
		return "", "", "", "",  "", "", s.logError(status.Error(codes.InvalidArgument,
			util.HostIpIsInvalid))
	}
	packageId = req.GetAppPackageId()
	if packageId == "" {
		return "", "", "", "",  "", "", s.logError(status.Error(codes.InvalidArgument,
			util.PackageIdIsInvalid))
	}
	tenantId = req.GetTenantId()
	err = util.ValidateUUID(tenantId)
	if err != nil {
		return "", "", "", "",  "", "", s.logError(status.Error(codes.InvalidArgument,
			util.TenantIdIsInvalid))
	}
	parameters := req.GetParameters()
	ak = parameters["ak"]
	if ak == "" {
		return "", "", "", "",  "", "", s.logError(status.Error(codes.InvalidArgument,
			util.AKIsInvalid))
	}
	sk = parameters["sk"]
	if sk == "" {
		return "", "", "", "",  "", "", s.logError(status.Error(codes.InvalidArgument,
			util.SKIsInvalid))
	}
	appInsId = req.GetAppInstanceId()
	err = util.ValidateUUID(appInsId)
	if err != nil {
		return "", "", "", "", "", "", s.logError(status.Error(codes.InvalidArgument,
			util.AppInsIdValid))
	}
	return tenantId, packageId, hostIp, appInsId, ak, sk, nil
}
func (s *ServerGRPC) validateInputParamsForUploadCfg(
	stream lcmservice.AppLCM_UploadConfigServer) (hostIp string, err error) {
	req, err := stream.Recv()
	if err != nil {
		log.Error(util.CannotReceivePackage)
		return
	}
	accessToken := req.GetAccessToken()
	err = util.ValidateAccessToken(accessToken, []string{util.MecmTenantRole, util.MecmAdminRole})
	if err != nil {
		if err.Error() == util.Forbidden {
			return "", s.logError(status.Error(codes.PermissionDenied, util.Forbidden))
		} else {
			return "", s.logError(status.Error(codes.InvalidArgument, util.AccssTokenIsInvalid))
		}
	}
	req, err = stream.Recv()
	if err != nil {
		log.Error(util.CannotReceivePackage)
		return
	}
	hostIp = req.GetHostIp()
	err = util.ValidateIpv4Address(hostIp)
	if err != nil {
		return "", s.logError(status.Error(codes.InvalidArgument, util.HostIpIsInvalid))
	}
	return hostIp, nil
}
func (s *ServerGRPC) validateInputParamsForPodDesc(
	req *lcmservice.WorkloadEventsRequest) (hostIp string, podName string, err error) {
	accessToken := req.GetAccessToken()
	err = util.ValidateAccessToken(accessToken, []string{util.MecmTenantRole, util.MecmAdminRole, util.MecmGuestRole})
	if err != nil {
		return "", "", s.logError(status.Error(codes.InvalidArgument,
			util.AccssTokenIsInvalid))
	}
	hostIp = req.GetHostIp()
	err = util.ValidateIpv4Address(hostIp)
	if err != nil {
		return "", "", s.logError(status.Error(codes.InvalidArgument, util.HostIpIsInvalid))
	}
	appInsId := req.GetAppInstanceId()
	err = util.ValidateUUID(appInsId)
	if err != nil {
		return "", "", s.logError(status.Error(codes.InvalidArgument, util.AppInsIdValid))
	}
	return hostIp, appInsId, nil
}