mirror of
https://github.com/labring/sealos.git
synced 2026-08-29 01:39:49 +08:00
Optimize billing efficiency (#5362)
* billing reconciler changed from controller trigger to scheduled task execution. * with manager runner * with manager runner * change cmv billing with manager runner * optimize log * fix reconcile * add billing with wait group goroutine * add logger * get the user cr using the cacheless client * optimize * Removes mongo Aggregate processing data that is too slow and uses in-memory batch user metering data. * fix * remove the owner that does not need to be updated * optimize log && fix the genGroupKey GenerateBillingDataFromRecords rules * make format
This commit is contained in:
Jiahui
@@ -59,6 +59,33 @@ import (
|
||||
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
|
||||
)
|
||||
|
||||
type CVMTaskRunner struct {
|
||||
DBClient database.Interface
|
||||
Logger logr.Logger
|
||||
*AccountReconciler
|
||||
}
|
||||
|
||||
func (r *CVMTaskRunner) Start(ctx context.Context) error {
|
||||
ticker := time.NewTicker(env.GetDurationEnvWithDefault("BILLING_CVM_INTERVAL", 10*time.Minute))
|
||||
defer func() {
|
||||
ticker.Stop()
|
||||
r.Logger.Info("stop billing cvm")
|
||||
}()
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
r.Logger.Info("start billing cvm", "time", time.Now().Format(time.RFC3339))
|
||||
err := r.BillingCVM()
|
||||
if err != nil {
|
||||
r.Logger.Error(err, "fail to billing cvm")
|
||||
}
|
||||
r.Logger.Info("end billing cvm", "time", time.Now().Format(time.RFC3339))
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const (
|
||||
ACCOUNTNAMESPACEENV = "ACCOUNT_NAMESPACE"
|
||||
DEFAULTACCOUNTNAMESPACE = "sealos-system"
|
||||
|
||||
@@ -22,27 +22,63 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"k8s.io/client-go/rest"
|
||||
|
||||
"k8s.io/client-go/kubernetes/scheme"
|
||||
|
||||
"github.com/labring/sealos/controllers/pkg/utils/env"
|
||||
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
|
||||
userv1 "github.com/labring/sealos/controllers/user/api/v1"
|
||||
|
||||
ctrl "sigs.k8s.io/controller-runtime"
|
||||
|
||||
"github.com/labring/sealos/controllers/pkg/types"
|
||||
|
||||
v12 "github.com/labring/sealos/controllers/account/api/v1"
|
||||
"github.com/labring/sealos/controllers/pkg/resources"
|
||||
|
||||
"sigs.k8s.io/controller-runtime/pkg/controller"
|
||||
|
||||
"github.com/go-logr/logr"
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
"sigs.k8s.io/controller-runtime/pkg/builder"
|
||||
"sigs.k8s.io/controller-runtime/pkg/event"
|
||||
"sigs.k8s.io/controller-runtime/pkg/predicate"
|
||||
|
||||
"github.com/labring/sealos/controllers/pkg/database"
|
||||
v1 "github.com/labring/sealos/controllers/user/api/v1"
|
||||
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
ctrl "sigs.k8s.io/controller-runtime"
|
||||
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||
)
|
||||
|
||||
type BillingTaskRunner struct {
|
||||
*BillingReconciler
|
||||
}
|
||||
|
||||
func (r *BillingTaskRunner) Start(ctx context.Context) error {
|
||||
if err := r.ExecuteBillingTask(); err != nil {
|
||||
r.Logger.Error(err, "failed to execute billing task")
|
||||
}
|
||||
defer func() {
|
||||
r.Logger.Info("stop billing reconcile", "time", time.Now().Format(time.RFC3339))
|
||||
}()
|
||||
now := time.Now()
|
||||
nextHour := now.Truncate(time.Hour).Add(time.Hour).Add(5 * time.Minute)
|
||||
r.Logger.Info("next billing reconcile time", "time", nextHour.Format(time.RFC3339))
|
||||
time.Sleep(nextHour.Sub(now))
|
||||
|
||||
ticker := time.NewTicker(time.Hour)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
if err := r.ExecuteBillingTask(); err != nil {
|
||||
r.Logger.Error(err, "failed to execute billing task")
|
||||
}
|
||||
select {
|
||||
case <-ticker.C:
|
||||
if err := r.ExecuteBillingTask(); err != nil {
|
||||
r.Logger.Error(err, "failed to execute billing task")
|
||||
}
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const (
|
||||
UserNamespacePrefix = "ns-"
|
||||
ResourceQuotaPrefix = "quota-"
|
||||
@@ -55,83 +91,110 @@ type BillingReconciler struct {
|
||||
client.Client
|
||||
Scheme *runtime.Scheme
|
||||
logr.Logger
|
||||
DBClient database.Account
|
||||
AccountV2 database.AccountV2
|
||||
Properties *resources.PropertyTypeLS
|
||||
DBClient database.Account
|
||||
AccountV2 database.AccountV2
|
||||
Properties *resources.PropertyTypeLS
|
||||
concurrentLimit int64
|
||||
}
|
||||
|
||||
//+kubebuilder:rbac:groups=core,resources=namespaces,verbs=get;list;watch;create;update;patch;delete
|
||||
//+kubebuilder:rbac:groups=core,resources=resourcequotas,verbs=get;list;watch;create;update;patch;delete
|
||||
//+kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=rolebindings,verbs=get;list;watch;create;update;patch;delete
|
||||
//+kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=roles,verbs=get;list;watch;create;update;patch;delete
|
||||
|
||||
// Reconcile is part of the main kubernetes reconciliation loop which aims to
|
||||
// move the current state of the cluster closer to the desired state.
|
||||
// TODO(user): Modify the Reconcile function to compare the state specified by
|
||||
// the Billing object against the actual cluster state, and then
|
||||
// perform operations to make the cluster state reflect the state specified by
|
||||
// the user.
|
||||
//
|
||||
// For more details, check Reconcile and its Result here:
|
||||
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.11.2/pkg/reconcile
|
||||
func (r *BillingReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
|
||||
r.Logger.V(1).Info("Reconcile Billing: ", "req.NamespacedName", req.NamespacedName)
|
||||
ns := &corev1.Namespace{}
|
||||
if err := r.Get(ctx, req.NamespacedName, ns); err != nil {
|
||||
return ctrl.Result{}, client.IgnoreNotFound(err)
|
||||
}
|
||||
|
||||
if ns.DeletionTimestamp != nil {
|
||||
r.Logger.V(1).Info("namespace is deleting", "namespace", ns)
|
||||
return ctrl.Result{}, nil
|
||||
}
|
||||
|
||||
owner := ns.Labels[v1.UserLabelOwnerKey]
|
||||
nsList, err := getOwnNsList(r.Client, owner)
|
||||
func (r *BillingReconciler) ExecuteBillingTask() error {
|
||||
r.Logger.Info("start billing reconcile", "time", time.Now().Format(time.RFC3339))
|
||||
ownerListMap, err := r.getRecentUsedOwners()
|
||||
if err != nil {
|
||||
r.Logger.Error(err, "get own namespace list failed")
|
||||
return ctrl.Result{Requeue: true}, err
|
||||
return fmt.Errorf("failed to get the owner list of the recently used resource: %w", err)
|
||||
}
|
||||
r.Logger.V(1).Info("own namespace list", "own", owner, "nsList", nsList)
|
||||
now := time.Now()
|
||||
currentHourTime := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.Local).UTC()
|
||||
queryTime := currentHourTime.Add(-1 * time.Hour)
|
||||
err = r.reconcileOwnerListBatch(ownerListMap, env.GetIntEnvWithDefault("BILLING_RECONCILE_BATCH_COUNT", 200), time.Now(), r.reconcileOwnerList)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to reconcile owner list batch: %w", err)
|
||||
}
|
||||
r.Logger.Info("finish billing reconcile", "time", time.Now().Format(time.RFC3339))
|
||||
return nil
|
||||
}
|
||||
|
||||
// TODO r.处理Unsettle状态的账单
|
||||
|
||||
if exist, lastUpdateTime, _ := r.DBClient.GetBillingLastUpdateTime(owner, v12.Consumption); exist {
|
||||
if lastUpdateTime.Equal(currentHourTime) || lastUpdateTime.After(currentHourTime) {
|
||||
return ctrl.Result{Requeue: true, RequeueAfter: time.Until(currentHourTime.Add(1*time.Hour + 10*time.Minute))}, nil
|
||||
}
|
||||
// 24小时内的数据，从上次更新时间开始计算，否则从当前时间起算
|
||||
if lastUpdateTime.After(currentHourTime.Add(-24 * time.Hour)) {
|
||||
queryTime = lastUpdateTime
|
||||
}
|
||||
func (r *BillingReconciler) reconcileOwnerList(ownerListMap map[string][]string, now time.Time) error {
|
||||
endHourTime := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.Local).UTC()
|
||||
startHourTime := endHourTime.Add(-1 * time.Hour)
|
||||
var ownerList, failedList []string
|
||||
for owner := range ownerListMap {
|
||||
ownerList = append(ownerList, owner)
|
||||
}
|
||||
updateOwnerList, err := r.DBClient.GetOwnersRecentUpdates(ownerList, endHourTime)
|
||||
if err != nil {
|
||||
return fmt.Errorf("get owners without recent updates failed: %w", err)
|
||||
}
|
||||
|
||||
orderList := []string{}
|
||||
consumAmount := int64(0)
|
||||
// 计算上次billing到当前的时间之间的整点，左开右闭
|
||||
for t := queryTime.Truncate(time.Hour).Add(time.Hour); t.Before(currentHourTime) || t.Equal(currentHourTime); t = t.Add(time.Hour) {
|
||||
ids, amount, err := r.DBClient.GenerateBillingData(t.Add(-1*time.Hour), t, r.Properties, nsList, getUsername(owner))
|
||||
if err != nil {
|
||||
return ctrl.Result{}, fmt.Errorf("generate billing data failed: %w", err)
|
||||
}
|
||||
orderList = append(orderList, ids...)
|
||||
consumAmount += amount
|
||||
// remove the owner that does not need to be updated
|
||||
for _, owner := range updateOwnerList {
|
||||
delete(ownerListMap, owner)
|
||||
}
|
||||
if consumAmount > 0 {
|
||||
if err := r.rechargeBalance(owner, consumAmount); err != nil {
|
||||
for i := range orderList {
|
||||
if err := r.DBClient.UpdateBillingStatus(orderList[i], resources.Unsettled); err != nil {
|
||||
r.Logger.Error(err, "update billing status failed", "id", orderList[i])
|
||||
}
|
||||
r.Logger.Info("get owners recent updates", "already update owner count", len(updateOwnerList), "remaining owner count", len(ownerListMap))
|
||||
|
||||
ownerBillings, err := r.DBClient.GenerateBillingData(startHourTime, endHourTime, r.Properties, ownerListMap)
|
||||
if err != nil {
|
||||
return fmt.Errorf("generate billing data failed: %w", err)
|
||||
}
|
||||
r.Logger.Info("generate billing data", "count", len(ownerBillings))
|
||||
for owner, billings := range ownerBillings {
|
||||
amount := int64(0)
|
||||
orderIDs := make([]string, 0, len(billings))
|
||||
for _, billing := range billings {
|
||||
amount += billing.Amount
|
||||
orderIDs = append(orderIDs, billing.OrderID)
|
||||
}
|
||||
if err = r.DBClient.SaveBillings(billings...); err != nil {
|
||||
r.Logger.Error(err, "save billings failed", "owner", owner, "amount", amount)
|
||||
failedList = append(failedList, owner)
|
||||
continue
|
||||
}
|
||||
if err := r.rechargeBalance(owner, amount); err != nil {
|
||||
r.Logger.Error(err, "recharge balance failed", "owner", owner, "amount", amount)
|
||||
failedList = append(failedList, owner)
|
||||
if err := r.DBClient.UpdateBillingStatus(orderIDs, resources.Unsettled); err != nil {
|
||||
r.Logger.Error(err, "update billing unsettled status failed", "orderIDs", orderIDs)
|
||||
}
|
||||
return ctrl.Result{}, fmt.Errorf("recharge balance failed: %w", err)
|
||||
}
|
||||
r.Logger.V(1).Info("success recharge balance", "owner", owner, "amount", consumAmount)
|
||||
}
|
||||
return ctrl.Result{Requeue: true, RequeueAfter: time.Until(currentHourTime.Add(1*time.Hour + 10*time.Minute))}, nil
|
||||
if len(failedList) > 0 {
|
||||
r.Logger.Error(fmt.Errorf("failed to reconcile owner list: %v", failedList), "failed to reconcile owner list")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// reconcileOwnerListBatch process ownerlistmap in batch mode
|
||||
func (r *BillingReconciler) reconcileOwnerListBatch(
|
||||
ownerListMap map[string][]string, // The owner -> namespaces mapping needs to be handled
|
||||
batchSize int, // number of owners processed per batch
|
||||
now time.Time, // current time
|
||||
reconcileFunc func(map[string][]string, time.Time) error, // processing function
|
||||
) error {
|
||||
if batchSize <= 0 {
|
||||
return fmt.Errorf("batch size must be greater than zero")
|
||||
}
|
||||
|
||||
owners := make([]string, 0, len(ownerListMap)) // store all owners
|
||||
for owner := range ownerListMap {
|
||||
owners = append(owners, owner)
|
||||
}
|
||||
|
||||
total := len(owners)
|
||||
for i := 0; i < total; i += batchSize {
|
||||
end := i + batchSize
|
||||
if end > total {
|
||||
end = total
|
||||
}
|
||||
|
||||
batchOwners := owners[i:end] // the owner list of the current batch
|
||||
batchOwnerMap := make(map[string][]string, len(batchOwners))
|
||||
for _, owner := range batchOwners {
|
||||
batchOwnerMap[owner] = ownerListMap[owner] // example retrieve a namespace
|
||||
}
|
||||
// call processing logic
|
||||
if err := reconcileFunc(batchOwnerMap, now); err != nil {
|
||||
return fmt.Errorf("failed to reconcile batch from %d to %d: %w", i, end, err)
|
||||
}
|
||||
r.Logger.Info("reconcile batch", "from", i, "to", end)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *BillingReconciler) rechargeBalance(owner string, amount int64) (err error) {
|
||||
@@ -144,48 +207,90 @@ func (r *BillingReconciler) rechargeBalance(owner string, amount int64) (err err
|
||||
return nil
|
||||
}
|
||||
|
||||
func getOwnNsList(clt client.Client, user string) ([]string, error) {
|
||||
nsList := &corev1.NamespaceList{}
|
||||
if err := clt.List(context.Background(), nsList, client.MatchingLabels{v1.UserLabelOwnerKey: user}); err != nil {
|
||||
return nil, fmt.Errorf("list namespace failed: %w", err)
|
||||
func (r *BillingReconciler) getRecentUsedOwners() (map[string][]string, error) {
|
||||
now := time.Now()
|
||||
endHourTime := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.Local).UTC()
|
||||
startHourTime := endHourTime.Add(-1 * time.Hour)
|
||||
namespaceList, err := r.DBClient.GetTimeUsedNamespaceList(startHourTime, endHourTime)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get recent owners failed: %w", err)
|
||||
}
|
||||
nsListStr := make([]string, len(nsList.Items))
|
||||
for i := range nsList.Items {
|
||||
nsListStr[i] = nsList.Items[i].Name
|
||||
nsToOwnerMap, err := GetAllUser()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get all user failed: %w", err)
|
||||
}
|
||||
return nsListStr, nil
|
||||
}
|
||||
|
||||
func (r *BillingReconciler) initDB() error {
|
||||
return r.DBClient.CreateBillingIfNotExist()
|
||||
}
|
||||
|
||||
// SetupWithManager sets up the controller with the Manager.
|
||||
func (r *BillingReconciler) SetupWithManager(mgr ctrl.Manager, rateOpts controller.Options) error {
|
||||
r.Logger = ctrl.Log.WithName("controller").WithName("Billing")
|
||||
if err := r.initDB(); err != nil {
|
||||
r.Logger.Error(err, "init db failed")
|
||||
r.Logger.Info("get owner and namespace", "owner count", len(nsToOwnerMap), "namespace count", len(namespaceList))
|
||||
usedOwnerList := make(map[string][]string)
|
||||
for _, ns := range namespaceList {
|
||||
if owner, ok := nsToOwnerMap[ns]; ok {
|
||||
if _, ok := usedOwnerList[owner]; !ok {
|
||||
usedOwnerList[owner] = []string{}
|
||||
}
|
||||
usedOwnerList[owner] = append(usedOwnerList[owner], ns)
|
||||
}
|
||||
}
|
||||
return ctrl.NewControllerManagedBy(mgr).
|
||||
For(&corev1.Namespace{}, builder.WithPredicates(predicate.Funcs{
|
||||
CreateFunc: func(createEvent event.CreateEvent) bool {
|
||||
own, ok := createEvent.Object.GetLabels()[v1.UserLabelOwnerKey]
|
||||
return ok && getUsername(createEvent.Object.GetName()) == own
|
||||
},
|
||||
UpdateFunc: func(_ event.UpdateEvent) bool {
|
||||
return false
|
||||
},
|
||||
DeleteFunc: func(_ event.DeleteEvent) bool {
|
||||
return false
|
||||
},
|
||||
GenericFunc: func(_ event.GenericEvent) bool {
|
||||
return false
|
||||
},
|
||||
})).
|
||||
WithOptions(rateOpts).
|
||||
Complete(r)
|
||||
r.Logger.Info("get all user", "count", len(usedOwnerList))
|
||||
return usedOwnerList, nil
|
||||
}
|
||||
|
||||
func getUsername(namespace string) string {
|
||||
return strings.TrimPrefix(namespace, UserNamespacePrefix)
|
||||
}
|
||||
|
||||
func (r *BillingReconciler) Init() error {
|
||||
r.Logger = ctrl.Log.WithName("controller").WithName("Billing")
|
||||
if err := r.DBClient.CreateBillingIfNotExist(); err != nil {
|
||||
return fmt.Errorf("create billing collection failed: %w", err)
|
||||
}
|
||||
r.concurrentLimit = env.GetInt64EnvWithDefault("BILLING_CONCURRENT_LIMIT", 100)
|
||||
return nil
|
||||
}
|
||||
|
||||
// map[namespace]owner
|
||||
func GetAllUser() (map[string]string, error) {
|
||||
err := userv1.AddToScheme(scheme.Scheme)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("unable to add scheme: %v", err)
|
||||
}
|
||||
config, err := rest.InClusterConfig()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("unable to build config: %v", err)
|
||||
}
|
||||
//TODO from cluster config
|
||||
//config, err := clientcmd.BuildConfigFromFlags("", os.Getenv("KUBECONFIG"))
|
||||
//if err != nil {
|
||||
// return nil, fmt.Errorf("unable to build config: %v", err)
|
||||
//}
|
||||
k8sClt, err := client.New(config, client.Options{Scheme: scheme.Scheme})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("unable to create client: %v", err)
|
||||
}
|
||||
nsToOwnerMap := make(map[string]string)
|
||||
|
||||
listOpts := &client.ListOptions{
|
||||
Limit: 5000,
|
||||
}
|
||||
for {
|
||||
userMetaList := &metav1.PartialObjectMetadataList{}
|
||||
userMetaList.SetGroupVersionKind(userv1.GroupVersion.WithKind("UserList"))
|
||||
|
||||
if err := k8sClt.List(context.Background(), userMetaList, listOpts); err != nil {
|
||||
return nil, fmt.Errorf("failed to list instances: %v", err)
|
||||
}
|
||||
|
||||
for _, user := range userMetaList.Items {
|
||||
owner := user.Annotations[userv1.UserLabelOwnerKey]
|
||||
if owner == "" {
|
||||
continue
|
||||
}
|
||||
nsToOwnerMap["ns-"+user.Name] = owner
|
||||
}
|
||||
|
||||
token := userMetaList.GetContinue()
|
||||
if token == "" {
|
||||
break
|
||||
}
|
||||
listOpts.Continue = token
|
||||
}
|
||||
return nsToOwnerMap, nil
|
||||
}
|
||||
|
||||
@@ -246,6 +246,18 @@ func (r *DebtReconciler) reconcile(ctx context.Context, userCr, userID string) e
|
||||
return nil
|
||||
}
|
||||
|
||||
func getOwnNsList(clt client.Client, user string) ([]string, error) {
|
||||
nsList := &corev1.NamespaceList{}
|
||||
if err := clt.List(context.Background(), nsList, client.MatchingLabels{userv1.UserLabelOwnerKey: user}); err != nil {
|
||||
return nil, fmt.Errorf("list namespace failed: %w", err)
|
||||
}
|
||||
nsListStr := make([]string, len(nsList.Items))
|
||||
for i := range nsList.Items {
|
||||
nsListStr[i] = nsList.Items[i].Name
|
||||
}
|
||||
return nsListStr, nil
|
||||
}
|
||||
|
||||
var ErrAccountNotExist = errors.New("account not exist")
|
||||
|
||||
/*
|
||||
|
||||
+36
-19
@@ -29,7 +29,6 @@ import (
|
||||
notificationv1 "github.com/labring/sealos/controllers/pkg/notification/api/v1"
|
||||
"github.com/labring/sealos/controllers/pkg/resources"
|
||||
"github.com/labring/sealos/controllers/pkg/types"
|
||||
"github.com/labring/sealos/controllers/pkg/utils/env"
|
||||
rate "github.com/labring/sealos/controllers/pkg/utils/rate"
|
||||
userv1 "github.com/labring/sealos/controllers/user/api/v1"
|
||||
|
||||
@@ -218,14 +217,23 @@ func main() {
|
||||
setupLog.Error(err, "unable to get property type")
|
||||
os.Exit(1)
|
||||
}
|
||||
if err = (&controllers.BillingReconciler{
|
||||
billingReconciler := controllers.BillingReconciler{
|
||||
DBClient: dbClient,
|
||||
Properties: resources.DefaultPropertyTypeLS,
|
||||
Client: mgr.GetClient(),
|
||||
Scheme: mgr.GetScheme(),
|
||||
AccountV2: v2Account,
|
||||
}).SetupWithManager(mgr, rateOpts); err != nil {
|
||||
setupManagerError(err, "Billing")
|
||||
}
|
||||
if err = billingReconciler.Init(); err != nil {
|
||||
setupLog.Error(err, "unable to init billing reconciler")
|
||||
os.Exit(1)
|
||||
}
|
||||
billingTaskRunner := &controllers.BillingTaskRunner{
|
||||
BillingReconciler: &billingReconciler,
|
||||
}
|
||||
if err := mgr.Add(billingTaskRunner); err != nil {
|
||||
setupLog.Error(err, "unable to add billing task runner")
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
if err = (&controllers.PodReconciler{
|
||||
@@ -261,23 +269,32 @@ func main() {
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
go func() {
|
||||
if cvmDBClient == nil {
|
||||
setupLog.Info("CVM DB client is nil, skip billing cvm")
|
||||
return
|
||||
if cvmDBClient != nil {
|
||||
cvmTaskRunner := &controllers.CVMTaskRunner{
|
||||
DBClient: cvmDBClient,
|
||||
Logger: ctrl.Log.WithName("CVMTaskRunner"),
|
||||
AccountReconciler: accountReconciler,
|
||||
}
|
||||
ticker := time.NewTicker(env.GetDurationEnvWithDefault("BILLING_CVM_INTERVAL", 10*time.Minute))
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
setupLog.Info("start billing cvm", "time", time.Now().Format(time.RFC3339))
|
||||
err := accountReconciler.BillingCVM()
|
||||
if err != nil {
|
||||
setupLog.Error(err, "fail to billing cvm")
|
||||
}
|
||||
setupLog.Info("end billing cvm", "time", time.Now().Format(time.RFC3339))
|
||||
<-ticker.C
|
||||
if err := mgr.Add(cvmTaskRunner); err != nil {
|
||||
setupLog.Error(err, "unable to add cvm task runner")
|
||||
os.Exit(1)
|
||||
}
|
||||
}()
|
||||
}
|
||||
//go func() {
|
||||
// now := time.Now()
|
||||
// nextHour := now.Truncate(time.Hour).Add(time.Hour)
|
||||
// time.Sleep(nextHour.Sub(now))
|
||||
//
|
||||
// ticker := time.NewTicker(time.Hour)
|
||||
// defer ticker.Stop()
|
||||
// for {
|
||||
// setupLog.Info("start billing reconcile", "time", time.Now().Format(time.RFC3339))
|
||||
// if err := billingReconciler.ExecuteBillingTask(); err != nil {
|
||||
// setupLog.Error(err, "failed to execute billing task")
|
||||
// }
|
||||
// <-ticker.C
|
||||
// }
|
||||
//}()
|
||||
|
||||
setupLog.Info("starting manager")
|
||||
if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
|
||||
|
||||
@@ -15,3 +15,8 @@
|
||||
package common
|
||||
|
||||
type Type int
|
||||
|
||||
const (
|
||||
Consumption Type = iota
|
||||
SubConsumption
|
||||
)
|
||||
|
||||
@@ -47,19 +47,21 @@ type CVM interface {
|
||||
|
||||
type Account interface {
|
||||
GetBillingLastUpdateTime(owner string, _type common.Type) (bool, time.Time, error)
|
||||
GetOwnersRecentUpdates(ownerList []string, checkTime time.Time) ([]string, error)
|
||||
GetTimeUsedNamespaceList(startTime, endTime time.Time) ([]string, error)
|
||||
SaveBillings(billing ...*resources.Billing) error
|
||||
SaveObjTraffic(obs ...*types.ObjectStorageTraffic) error
|
||||
GetAllLatestObjTraffic(startTime, endTime time.Time) ([]types.ObjectStorageTraffic, error)
|
||||
HandlerTimeObjBucketSentTraffic(startTime, endTime time.Time, bucket string) (int64, error)
|
||||
GetTimeObjBucketBucket(startTime, endTime time.Time) ([]string, error)
|
||||
GetUnsettingBillingHandler(owner string) ([]resources.BillingHandler, error)
|
||||
UpdateBillingStatus(orderID string, status resources.BillingStatus) error
|
||||
UpdateBillingStatus(orderIDs []string, status resources.BillingStatus) error
|
||||
GetUpdateTimeForCategoryAndPropertyFromMetering(category string, property string) (time.Time, error)
|
||||
GetAllPayment() ([]resources.Billing, error)
|
||||
InitDefaultPropertyTypeLS() error
|
||||
SavePropertyTypes(types []resources.PropertyType) error
|
||||
GetBillingCount(accountType common.Type, startTime, endTime time.Time) (count, amount int64, err error)
|
||||
GenerateBillingData(startTime, endTime time.Time, prols *resources.PropertyTypeLS, namespaces []string, owner string) (orderID []string, amount int64, err error)
|
||||
GenerateBillingData(startTime, endTime time.Time, prols *resources.PropertyTypeLS, ownerToNS map[string][]string) (map[string][]*resources.Billing, error)
|
||||
InsertMonitor(ctx context.Context, monitors ...*resources.Monitor) error
|
||||
GetDistinctMonitorCombinations(startTime, endTime time.Time) ([]resources.Monitor, error)
|
||||
DropMonitorCollectionsOlderThan(days int) error
|
||||
|
||||
@@ -149,6 +149,87 @@ func (m *mongoDB) GetBillingLastUpdateTime(owner string, _type common.Type) (boo
|
||||
return false, time.Time{}, fmt.Errorf("failed to convert time field to primitive.DateTime: %v", result["time"])
|
||||
}
|
||||
|
||||
func (m *mongoDB) GetOwnersRecentUpdates(ownerList []string, checkTime time.Time) ([]string, error) {
|
||||
// MongoDB filter
|
||||
filter := bson.M{
|
||||
"owner": bson.M{"$in": ownerList},
|
||||
"type": common.Consumption,
|
||||
"app_type": bson.M{
|
||||
"$ne": resources.AppType[resources.CVM],
|
||||
},
|
||||
}
|
||||
|
||||
// Aggregate query: Group by owner to get the latest time
|
||||
pipeline := mongo.Pipeline{
|
||||
{{Key: "$match", Value: filter}},
|
||||
{{Key: "$sort", Value: bson.D{{Key: "owner", Value: 1}, {Key: "time", Value: -1}}}}, // Sort by owner first, then by time in descending order
|
||||
{{Key: "$group", Value: bson.D{
|
||||
{Key: "_id", Value: "$owner"},
|
||||
{Key: "lastUpdateTime", Value: bson.D{{Key: "$first", Value: "$time"}}}, // fetch latest Time
|
||||
}}},
|
||||
}
|
||||
// execute aggregate query
|
||||
cursor, err := m.getBillingCollection().Aggregate(context.Background(), pipeline)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to execute aggregate query: %w", err)
|
||||
}
|
||||
defer cursor.Close(context.Background())
|
||||
|
||||
// get all the data out at once
|
||||
var results []struct {
|
||||
Owner string `bson:"_id"`
|
||||
LastUpdateRaw primitive.DateTime `bson:"lastUpdateTime"`
|
||||
}
|
||||
if err := cursor.All(context.Background(), &results); err != nil {
|
||||
return nil, fmt.Errorf("failed to decode cursor: %w", err)
|
||||
}
|
||||
|
||||
// use map to store query results
|
||||
latestUpdates := make(map[string]time.Time, len(results))
|
||||
for _, result := range results {
|
||||
latestUpdates[result.Owner] = result.LastUpdateRaw.Time()
|
||||
}
|
||||
|
||||
// **In-memory processing: Filters owners that have been updated since checkTime**
|
||||
var updatedOwners []string
|
||||
for _, owner := range ownerList {
|
||||
lastUpdateTime, exists := latestUpdates[owner]
|
||||
if exists && (lastUpdateTime.After(checkTime) || lastUpdateTime.Equal(checkTime)) {
|
||||
updatedOwners = append(updatedOwners, owner)
|
||||
}
|
||||
}
|
||||
return updatedOwners, nil
|
||||
}
|
||||
|
||||
func (m *mongoDB) GetTimeUsedNamespaceList(startTime, endTime time.Time) ([]string, error) {
|
||||
pipeline := mongo.Pipeline{
|
||||
{{Key: "$match", Value: bson.D{{Key: "time", Value: bson.D{{Key: "$gte", Value: startTime}, {Key: "$lt", Value: endTime}}}}}},
|
||||
{{Key: "$group", Value: bson.D{{Key: "_id", Value: "$category"}}}},
|
||||
{{Key: "$project", Value: bson.D{{Key: "_id", Value: 0}, {Key: "namespace", Value: "$_id"}}}},
|
||||
}
|
||||
cursor, err := m.getMonitorCollection(startTime).Aggregate(context.Background(), pipeline)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("aggregate error: %v", err)
|
||||
}
|
||||
defer cursor.Close(context.Background())
|
||||
|
||||
var namespaces []string
|
||||
for cursor.Next(context.Background()) {
|
||||
var result struct {
|
||||
Namespace string `bson:"namespace"`
|
||||
}
|
||||
err := cursor.Decode(&result)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("decode error: %v", err)
|
||||
}
|
||||
namespaces = append(namespaces, result.Namespace)
|
||||
}
|
||||
if err = cursor.Err(); err != nil {
|
||||
return nil, fmt.Errorf("cursor error: %v", err)
|
||||
}
|
||||
return namespaces, nil
|
||||
}
|
||||
|
||||
func (m *mongoDB) GetUnsettingBillingHandler(owner string) ([]resources.BillingHandler, error) {
|
||||
filter := bson.M{
|
||||
"owner": owner,
|
||||
@@ -178,15 +259,15 @@ func (m *mongoDB) GetUnsettingBillingHandler(owner string) ([]resources.BillingH
|
||||
return results, nil
|
||||
}
|
||||
|
||||
func (m *mongoDB) UpdateBillingStatus(orderID string, status resources.BillingStatus) error {
|
||||
func (m *mongoDB) UpdateBillingStatus(orderIDs []string, status resources.BillingStatus) error {
|
||||
// create a query filter
|
||||
filter := bson.M{"order_id": orderID}
|
||||
filter := bson.M{"order_id": bson.M{"$in": orderIDs}}
|
||||
update := bson.M{
|
||||
"$set": bson.M{
|
||||
"status": status,
|
||||
},
|
||||
}
|
||||
_, err := m.getBillingCollection().UpdateOne(context.Background(), filter, update)
|
||||
_, err := m.getBillingCollection().UpdateMany(context.Background(), filter, update)
|
||||
if err != nil {
|
||||
return fmt.Errorf("update error: %v", err)
|
||||
}
|
||||
@@ -421,185 +502,234 @@ func (m *mongoDB) SavePropertyTypes(types []resources.PropertyType) error {
|
||||
return err
|
||||
}
|
||||
|
||||
func (m *mongoDB) GenerateBillingData(startTime, endTime time.Time, prols *resources.PropertyTypeLS, namespaces []string, owner string) (orderID []string, amount int64, err error) {
|
||||
minutes := endTime.Sub(startTime).Minutes()
|
||||
|
||||
groupStage := bson.D{
|
||||
primitive.E{Key: "_id", Value: bson.D{{Key: "type", Value: "$type"}, {Key: "name", Value: "$name"}, {Key: "category", Value: "$category"}, {Key: "parent_type", Value: "$parent_type"}, {Key: "parent_name", Value: "$parent_name"}}},
|
||||
primitive.E{Key: "count", Value: bson.D{{Key: "$sum", Value: 1}}},
|
||||
}
|
||||
|
||||
projectStage := bson.D{
|
||||
primitive.E{Key: "_id", Value: 0},
|
||||
primitive.E{Key: "type", Value: "$_id.type"},
|
||||
primitive.E{Key: "name", Value: "$_id.name"},
|
||||
primitive.E{Key: "parent_type", Value: "$_id.parent_type"},
|
||||
primitive.E{Key: "parent_name", Value: "$_id.parent_name"},
|
||||
primitive.E{Key: "category", Value: "$_id.category"},
|
||||
}
|
||||
|
||||
// initialize the used phase
|
||||
usedStage := bson.M{}
|
||||
|
||||
// Build the $group and $project phases dynamically from EnumMap
|
||||
for key, value := range prols.EnumMap {
|
||||
keyStr := strconv.Itoa(int(key))
|
||||
|
||||
// $max - $min;
|
||||
// When max is not zero, the minimum value other than the zero value is used to prevent some data from obtaining a value in special cases
|
||||
// max-min=0 if the hour has only one data piece or no data piece
|
||||
if value.PriceType == resources.DIF {
|
||||
// for non 0 $min
|
||||
minWithCondition := bson.D{
|
||||
{Key: "$min", Value: bson.D{
|
||||
{Key: "$cond", Value: bson.A{
|
||||
bson.D{{Key: "$eq", Value: bson.A{"$used." + keyStr, 0}}},
|
||||
nil, // 将0值排除在外
|
||||
"$used." + keyStr,
|
||||
}},
|
||||
}},
|
||||
}
|
||||
|
||||
groupStage = append(groupStage,
|
||||
primitive.E{Key: keyStr + "_max", Value: bson.D{{Key: "$max", Value: "$used." + keyStr}}}, // 正常计算$max
|
||||
primitive.E{Key: keyStr + "_min", Value: minWithCondition},
|
||||
)
|
||||
|
||||
// added to the used phase
|
||||
usedStage[keyStr] = bson.D{{Key: "$subtract", Value: bson.A{
|
||||
"$" + keyStr + "_max",
|
||||
"$" + keyStr + "_min",
|
||||
}}}
|
||||
continue
|
||||
}
|
||||
if value.PriceType == resources.SUM {
|
||||
groupStage = append(groupStage, primitive.E{Key: keyStr, Value: bson.D{{Key: "$sum", Value: "$used." + keyStr}}})
|
||||
usedStage[keyStr] = bson.D{{Key: "$toInt", Value: "$" + keyStr}}
|
||||
continue
|
||||
}
|
||||
groupStage = append(groupStage, primitive.E{Key: keyStr, Value: bson.D{{Key: "$sum", Value: "$used." + keyStr}}})
|
||||
usedStage[keyStr] = bson.D{{Key: "$toInt", Value: bson.D{{Key: "$round", Value: bson.D{{Key: "$divide", Value: bson.A{
|
||||
"$" + keyStr, minutes}}}}}}}
|
||||
}
|
||||
|
||||
// add the used phase to the $project phase
|
||||
projectStage = append(projectStage, primitive.E{Key: "used", Value: usedStage})
|
||||
|
||||
// construction-pipeline
|
||||
pipeline := mongo.Pipeline{
|
||||
{{Key: "$match", Value: bson.D{{Key: "time", Value: bson.D{{Key: "$gte", Value: startTime}, {Key: "$lt", Value: endTime}}}, {Key: "category", Value: bson.D{{Key: "$in", Value: namespaces}}}}}},
|
||||
{{Key: "$group", Value: groupStage}},
|
||||
{{Key: "$project", Value: projectStage}},
|
||||
}
|
||||
|
||||
cursor, err := m.getMonitorCollection(startTime).Aggregate(context.Background(), pipeline)
|
||||
func (m *mongoDB) GenerateBillingData(startTime, endTime time.Time, prols *resources.PropertyTypeLS, ownerToNS map[string][]string) (map[string][]*resources.Billing, error) {
|
||||
ownerMonitors, err := m.FetchOwnerMonitorRecords(startTime, endTime, ownerToNS)
|
||||
if err != nil {
|
||||
return nil, 0, fmt.Errorf("aggregate error: %v", err)
|
||||
return nil, fmt.Errorf("failed to fetch monitor records: %v", err)
|
||||
}
|
||||
var ownerBillings = make(map[string][]*resources.Billing)
|
||||
for owner, monitors := range ownerMonitors {
|
||||
billings, err := GenerateBillingDataFromRecords(monitors, prols, startTime, endTime, owner)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to generate billing data: %v", err)
|
||||
}
|
||||
ownerBillings[owner] = billings
|
||||
}
|
||||
return ownerBillings, nil
|
||||
}
|
||||
|
||||
func (m *mongoDB) FetchOwnerMonitorRecords(startTime, endTime time.Time, ownerToNS map[string][]string) (map[string][]resources.Monitor, error) {
|
||||
// collect all namespaces to avoid repetition
|
||||
nsSet := make(map[string]struct{})
|
||||
for _, nsList := range ownerToNS {
|
||||
for _, ns := range nsList {
|
||||
nsSet[ns] = struct{}{}
|
||||
}
|
||||
}
|
||||
namespaces := make([]string, 0, len(nsSet))
|
||||
for ns := range nsSet {
|
||||
namespaces = append(namespaces, ns)
|
||||
}
|
||||
|
||||
// get all matching monitor records from mongodb
|
||||
collection := m.getMonitorCollection(startTime)
|
||||
filter := bson.M{
|
||||
"time": bson.M{"$gte": startTime, "$lt": endTime},
|
||||
"category": bson.M{"$in": namespaces}, // 查询所有涉及的 namespaces
|
||||
}
|
||||
|
||||
cursor, err := collection.Find(context.Background(), filter)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to find monitor records: %w", err)
|
||||
}
|
||||
defer cursor.Close(context.Background())
|
||||
|
||||
var appCostsMap = make(map[string]map[string][]resources.AppCost)
|
||||
// map[ns/type]int64
|
||||
var nsTypeAmount = make(map[string]int64)
|
||||
// reading mongodb data
|
||||
var allRecords []resources.Monitor
|
||||
if err := cursor.All(context.Background(), &allRecords); err != nil {
|
||||
return nil, fmt.Errorf("failed to decode monitor records: %w", err)
|
||||
}
|
||||
|
||||
for cursor.Next(context.Background()) {
|
||||
var result struct {
|
||||
Type uint8 `bson:"type"`
|
||||
Namespace string `bson:"category"`
|
||||
Name string `bson:"name"`
|
||||
ParentType uint8 `bson:"parent_type"`
|
||||
ParentName string `bson:"parent_name"`
|
||||
Used resources.EnumUsedMap `bson:"used"`
|
||||
}
|
||||
|
||||
err := cursor.Decode(&result)
|
||||
if err != nil {
|
||||
return nil, 0, fmt.Errorf("decode error: %v", err)
|
||||
}
|
||||
|
||||
//TODO delete
|
||||
//logger.Info("generate billing data", "result", result)
|
||||
|
||||
if _, ok := appCostsMap[result.Namespace]; !ok {
|
||||
appCostsMap[result.Namespace] = make(map[string][]resources.AppCost)
|
||||
}
|
||||
strType := strconv.Itoa(int(result.Type))
|
||||
if _, ok := appCostsMap[result.Namespace][strType]; !ok {
|
||||
appCostsMap[result.Namespace][strType] = make([]resources.AppCost, 0)
|
||||
}
|
||||
appCost := resources.AppCost{
|
||||
Type: result.Type,
|
||||
Used: result.Used,
|
||||
Name: result.Name,
|
||||
UsedAmount: make(map[uint8]int64),
|
||||
}
|
||||
// Calculate the amount and set the used value
|
||||
for property := range result.Used {
|
||||
if prop, ok := prols.EnumMap[property]; ok {
|
||||
if prop.UnitPrice > 0 {
|
||||
appCost.UsedAmount[property] = int64(math.Ceil(float64(result.Used[property]) * prop.UnitPrice))
|
||||
appCost.Amount += appCost.UsedAmount[property]
|
||||
// build the mapping of owner monitor data
|
||||
ownerMonitorRecords := make(map[string][]resources.Monitor)
|
||||
for _, record := range allRecords {
|
||||
for owner, nsList := range ownerToNS {
|
||||
// Only the records of the namespace that belong to the owner are saved
|
||||
for _, ns := range nsList {
|
||||
if record.Category == ns {
|
||||
ownerMonitorRecords[owner] = append(ownerMonitorRecords[owner], record)
|
||||
break // avoid duplicate additions
|
||||
}
|
||||
}
|
||||
}
|
||||
if appCost.Amount == 0 {
|
||||
continue
|
||||
}
|
||||
key := result.Namespace + "/" + strType
|
||||
if result.ParentType != 0 && result.ParentName != "" {
|
||||
key = result.Namespace + "/" + strconv.Itoa(int(result.ParentType)) + "/" + result.ParentName
|
||||
}
|
||||
nsTypeAmount[key] += appCost.Amount
|
||||
appCostsMap[result.Namespace][key] = append(appCostsMap[result.Namespace][key], appCost)
|
||||
}
|
||||
|
||||
return ownerMonitorRecords, nil
|
||||
}
|
||||
|
||||
func GenerateBillingDataFromRecords(records []resources.Monitor, prols *resources.PropertyTypeLS, startTime, endTime time.Time, owner string) (billings []*resources.Billing, err error) {
|
||||
// Calculate the interval (minutes) to ensure that the divisor is not 0
|
||||
minutes := math.Max(endTime.Sub(startTime).Minutes(), 1)
|
||||
|
||||
// 存储分组后的数据
|
||||
aggregatedMap := make(map[string]*struct {
|
||||
resources.Monitor
|
||||
UsedValues map[uint8][]int64
|
||||
Count int64
|
||||
})
|
||||
|
||||
// 分组 key 生成规则
|
||||
genGroupKey := func(rec resources.Monitor) string {
|
||||
return fmt.Sprintf("%s/%d/%s", rec.Category, rec.Type, rec.Name)
|
||||
}
|
||||
|
||||
// 遍历所有记录，按分组键聚合
|
||||
for _, rec := range records {
|
||||
key := genGroupKey(rec)
|
||||
if _, ok := aggregatedMap[key]; !ok {
|
||||
aggregatedMap[key] = &struct {
|
||||
resources.Monitor
|
||||
UsedValues map[uint8][]int64
|
||||
Count int64
|
||||
}{
|
||||
Monitor: rec,
|
||||
UsedValues: make(map[uint8][]int64),
|
||||
Count: 0,
|
||||
}
|
||||
}
|
||||
aggregatedMap[key].Count++
|
||||
for k, v := range rec.Used {
|
||||
//aggregatedMap[key].UsedValues[k] = append(aggregatedMap[key].UsedValues[k], v)
|
||||
if _, exists := aggregatedMap[key].UsedValues[k]; !exists {
|
||||
aggregatedMap[key].UsedValues[k] = make([]int64, 0, len(records))
|
||||
}
|
||||
aggregatedMap[key].UsedValues[k] = append(aggregatedMap[key].UsedValues[k], v)
|
||||
}
|
||||
}
|
||||
|
||||
// 存储最终计费数据
|
||||
// map[namespace]map[app_type | parent_type/parent_name][]resources.AppCost
|
||||
appCostsMap := make(map[string]map[string][]resources.AppCost)
|
||||
nsTypeAmount := make(map[string]map[string]int64)
|
||||
|
||||
calculateFinalUsed := func(values map[uint8][]int64, prols *resources.PropertyTypeLS, minutes float64) map[uint8]int64 {
|
||||
finalUsed := make(map[uint8]int64)
|
||||
for propKey, vals := range values {
|
||||
if prop, ok := prols.EnumMap[propKey]; ok {
|
||||
finalUsed[propKey] = computeUsedValue(vals, prop, minutes)
|
||||
}
|
||||
}
|
||||
return finalUsed
|
||||
}
|
||||
// 计算最终 Used 数据
|
||||
for _, agg := range aggregatedMap {
|
||||
finalUsed := calculateFinalUsed(agg.UsedValues, prols, minutes)
|
||||
// 计算费用
|
||||
appCost := resources.AppCost{
|
||||
Type: agg.Type,
|
||||
Name: agg.Name,
|
||||
Used: finalUsed,
|
||||
UsedAmount: make(map[uint8]int64),
|
||||
}
|
||||
var totalAmount int64
|
||||
for propKey, usedVal := range finalUsed {
|
||||
if prop, ok := prols.EnumMap[propKey]; ok {
|
||||
if prop.UnitPrice > 0 {
|
||||
feeFloat := float64(usedVal) * prop.UnitPrice
|
||||
if feeFloat > math.MaxInt64 {
|
||||
return nil, fmt.Errorf("fee calculation overflow: %f", feeFloat)
|
||||
}
|
||||
fee := int64(math.Ceil(feeFloat))
|
||||
appCost.UsedAmount[propKey] = fee
|
||||
totalAmount += fee
|
||||
}
|
||||
}
|
||||
}
|
||||
if totalAmount == 0 {
|
||||
continue
|
||||
}
|
||||
appCost.Amount = totalAmount
|
||||
groupKey := strconv.Itoa(int(agg.Type))
|
||||
if agg.ParentType != 0 && agg.ParentName != "" {
|
||||
groupKey = strconv.Itoa(int(agg.ParentType)) + "/" + agg.ParentName
|
||||
}
|
||||
ns := agg.Category
|
||||
if _, ok := nsTypeAmount[ns]; !ok {
|
||||
nsTypeAmount[ns] = make(map[string]int64)
|
||||
}
|
||||
nsTypeAmount[ns][groupKey] += totalAmount
|
||||
|
||||
if _, ok := appCostsMap[ns]; !ok {
|
||||
appCostsMap[ns] = make(map[string][]resources.AppCost)
|
||||
}
|
||||
appCostsMap[ns][groupKey] = append(appCostsMap[ns][groupKey], appCost)
|
||||
}
|
||||
billings = make([]*resources.Billing, 0)
|
||||
|
||||
// 生成 Billing 数据
|
||||
for ns, appCostMap := range appCostsMap {
|
||||
for tp, appCost := range appCostMap {
|
||||
amountt := nsTypeAmount[tp]
|
||||
if amountt == 0 {
|
||||
for tp, appCostList := range appCostMap {
|
||||
amount := nsTypeAmount[ns][tp]
|
||||
if amount <= 0 {
|
||||
continue
|
||||
}
|
||||
id, err := gonanoid.New(12)
|
||||
if err != nil {
|
||||
return nil, 0, fmt.Errorf("generate billing id error: %v", err)
|
||||
return nil, fmt.Errorf("generate billing id error: %v", err)
|
||||
}
|
||||
// tp = ns/type/parentName && parentName not contain "/"
|
||||
appType, appName := 0, ""
|
||||
switch strings.Count(tp, "/") {
|
||||
case 1:
|
||||
appType, _ = strconv.Atoi(strings.Split(tp, "/")[1])
|
||||
case 2:
|
||||
appType, _ = strconv.Atoi(strings.Split(tp, "/")[1])
|
||||
appName = strings.Split(tp, "/")[2]
|
||||
parts := strings.Split(tp, "/")
|
||||
appType, _ := strconv.Atoi(parts[0])
|
||||
appName := ""
|
||||
if len(parts) > 1 {
|
||||
appName = parts[1]
|
||||
}
|
||||
billing := resources.Billing{
|
||||
billings = append(billings, &resources.Billing{
|
||||
OrderID: id,
|
||||
Type: Consumption,
|
||||
Namespace: ns,
|
||||
AppType: uint8(appType),
|
||||
AppName: appName,
|
||||
AppCosts: appCost,
|
||||
Amount: amountt,
|
||||
AppCosts: appCostList,
|
||||
Amount: amount,
|
||||
Owner: owner,
|
||||
Time: endTime,
|
||||
Status: resources.Settled,
|
||||
}
|
||||
amount += amountt
|
||||
orderID = append(orderID, id)
|
||||
// Insert the billing document
|
||||
_, err = m.getBillingCollection().InsertOne(context.Background(), billing)
|
||||
if err != nil {
|
||||
return nil, 0, fmt.Errorf("insert error: %v", err)
|
||||
}
|
||||
//TODO delete
|
||||
//logger.Info("generate billing data", "billing", billing)
|
||||
})
|
||||
}
|
||||
}
|
||||
return billings, nil
|
||||
}
|
||||
|
||||
if err = cursor.Err(); err != nil {
|
||||
return nil, 0, fmt.Errorf("cursor error: %v", err)
|
||||
func computeUsedValue(usedValues []int64, prop resources.PropertyType, minutes float64) int64 {
|
||||
switch prop.PriceType {
|
||||
case resources.DIF:
|
||||
var maxVal int64 = -math.MaxInt64
|
||||
var minVal int64 = math.MaxInt64
|
||||
for _, v := range usedValues {
|
||||
if v > maxVal {
|
||||
maxVal = v
|
||||
}
|
||||
if v != 0 && v < minVal {
|
||||
minVal = v
|
||||
}
|
||||
}
|
||||
if maxVal > minVal {
|
||||
return maxVal - minVal
|
||||
}
|
||||
return 0
|
||||
case resources.SUM:
|
||||
var sum int64
|
||||
for _, v := range usedValues {
|
||||
sum += v
|
||||
}
|
||||
return sum
|
||||
default:
|
||||
var sum int64
|
||||
for _, v := range usedValues {
|
||||
sum += v
|
||||
}
|
||||
return int64(math.Round(float64(sum) / minutes))
|
||||
}
|
||||
return orderID, amount, nil
|
||||
}
|
||||
|
||||
func (m *mongoDB) GetUpdateTimeForCategoryAndPropertyFromMetering(category string, property string) (time.Time, error) {
|
||||
|
||||
@@ -294,27 +294,6 @@ info generate billing data used {2 ns-7uyfrr47 pay-xy map[0:325 1:166 2:0]}
|
||||
cpu: 500m
|
||||
memory: 256Mi
|
||||
*/
|
||||
func TestMongoDB_GenerateBillingData(t *testing.T) {
|
||||
dbCTX := context.Background()
|
||||
|
||||
m, err := NewMongoInterface(dbCTX, os.Getenv("MONGODB_URI"))
|
||||
if err != nil {
|
||||
t.Errorf("failed to connect mongo: error = %v", err)
|
||||
}
|
||||
defer func() {
|
||||
if err = m.Disconnect(dbCTX); err != nil {
|
||||
t.Errorf("failed to disconnect mongo: error = %v", err)
|
||||
}
|
||||
}()
|
||||
queryTime := time.Now().UTC()
|
||||
|
||||
ids, amount, err := m.GenerateBillingData(queryTime.Add(-1*time.Hour), queryTime, resources.DefaultPropertyTypeLS, []string{"ns-7uyfrr47", "ns-1jc12uh6", "ns-ezplle8l"}, "1jc12uh6")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to generate billing data: %v", err)
|
||||
}
|
||||
t.Logf("generate billing data used %v", amount)
|
||||
t.Logf("generate billing data used %v", ids)
|
||||
}
|
||||
|
||||
func TestMongoDB_SetPropertyTypeLS(t *testing.T) {
|
||||
dbCTX := context.Background()
|
||||
@@ -479,3 +458,72 @@ func Test_mongoDB_GetTimeObjBucketBucket(t *testing.T) {
|
||||
t.Logf("bucket: %#+v", bucket)
|
||||
}
|
||||
}
|
||||
|
||||
func Test_mongoDB_GetTimeUsedOwnerList(t *testing.T) {
|
||||
dbCTX := context.Background()
|
||||
|
||||
m, err := NewMongoInterface(dbCTX, "")
|
||||
if err != nil {
|
||||
t.Errorf("failed to connect mongo: error = %v", err)
|
||||
}
|
||||
defer func() {
|
||||
if err = m.Disconnect(dbCTX); err != nil {
|
||||
t.Errorf("failed to disconnect mongo: error = %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
owners, err := m.GetTimeUsedNamespaceList(time.Now().UTC().Add(-time.Hour), time.Now().UTC())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to get time used owner list: %v", err)
|
||||
}
|
||||
t.Logf("get time used owner list success: %v", owners)
|
||||
}
|
||||
|
||||
func Test_mongoDB_GenerateBillingData(t *testing.T) {
|
||||
dbCTX := context.Background()
|
||||
|
||||
m, err := NewMongoInterface(dbCTX, os.Getenv("MONGO_URI"))
|
||||
if err != nil {
|
||||
t.Errorf("failed to connect mongo: error = %v", err)
|
||||
}
|
||||
defer func() {
|
||||
if err = m.Disconnect(dbCTX); err != nil {
|
||||
t.Errorf("failed to disconnect mongo: error = %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
prols := resources.DefaultPropertyTypeLS
|
||||
ownerToNS := map[string][]string{
|
||||
"ax1uut8w": {"ns-tnw80mhk", "ns-ax1uut8w"},
|
||||
}
|
||||
billings, err := m.GenerateBillingData(time.Now().UTC().Add(-time.Hour), time.Now().UTC(), prols, ownerToNS)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to generate billing data: %v", err)
|
||||
}
|
||||
for _, billing := range billings {
|
||||
for _, bill := range billing {
|
||||
t.Logf("%+v\n", bill)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func Test_mongoDB_GetOwnersWithoutRecentUpdates(t *testing.T) {
|
||||
dbCTX := context.Background()
|
||||
|
||||
m, err := NewMongoInterface(dbCTX, "")
|
||||
if err != nil {
|
||||
t.Errorf("failed to connect mongo: error = %v", err)
|
||||
}
|
||||
defer func() {
|
||||
if err = m.Disconnect(dbCTX); err != nil {
|
||||
t.Errorf("failed to disconnect mongo: error = %v", err)
|
||||
}
|
||||
}()
|
||||
now := time.Now().UTC()
|
||||
endHourTime := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.Local).UTC()
|
||||
owners, err := m.GetOwnersRecentUpdates([]string{"nfhmc74p"}, endHourTime)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to get owners without recent updates: %v", err)
|
||||
}
|
||||
t.Logf("get owners without recent updates success: %v", owners)
|
||||
}
|
||||
|
||||
@@ -135,7 +135,7 @@ type Billing struct {
|
||||
// if type = Transfer, then transfer is not nil
|
||||
Transfer *Transfer `json:"transfer" bson:"transfer,omitempty"`
|
||||
Detail string `json:"detail" bson:"detail,omitempty"`
|
||||
UserUID uuid.UUID `json:"user_uid" bson:"user_uid,omitempty"`
|
||||
//UserUID uuid.UUID `json:"user_uid" bson:"user_uid,omitempty"`
|
||||
}
|
||||
|
||||
type Payment struct {
|
||||
|
||||
Reference in New Issue
Block a user