odbtest/test/client.go

package main

import (
	"fmt"
	"os"
	"runtime"
	"strconv"
	"strings"
	"time"

	"git.wecise.com/wecise/odb-go/odb"
	"git.wecise.com/wecise/odbserver/lib/wgwaiter"
	"gitee.com/wecisecode/util/logger"
	"gitee.com/wecisecode/util/set/strset"
	//"git.wecise.com/wecise/odbserver/odb/upsertupsert"
)

var odbcfg *odb.Config

func init() {
	odbpath := os.Getenv("ODBPATH")
	if odbpath == "" {
		odbpath = "47.92.151.165:11001"
	}
	odbpaths := strset.New(strings.Split(odbpath, ",")...).List()
	keyspace := os.Getenv("KEYSPACE")
	if keyspace == "" {
		keyspace = "ooootest"
	}

	odbcfg = &odb.Config{
		Hosts:    odbpaths,
		Keyspace: keyspace,
		User:     fmt.Sprint("测试客户端"),
		Pass:     "********",
		PoolSize: 20,
		Debug:    true,
	}
}

func getClient() (c odb.Client, err error) {
	client, err := odb.NewClient(odbcfg)
	if err != nil {
		return nil, err
	}
	return client, nil
}

func main() {
	time.Sleep(2 * time.Second)

	client, err := getClient()
	if err != nil {
		logger.Error(err)
	}
	defer func() {
		if err := client.Close(); err != nil {
			logger.Error(err)
		}
	}()

	wgg := wgwaiter.NewWaiter()
	DEFAULT_MQL_CONCURRENT_NUM := runtime.NumCPU()

	for kk := 0; kk < 10; kk++ {

		for i := kk * DEFAULT_MQL_CONCURRENT_NUM; i < (kk+1)*DEFAULT_MQL_CONCURRENT_NUM; i++ {
			wgg.AddOne()
			go func(ii int) {

				mtype := ii % 10

				tag := "mytag"
				switch mtype {
				case 1:
					tag = "mytag1"
				case 2:
					tag = "mytag2"
				case 3:
					tag = "mytag3"
				case 4:
					tag = "mytag4"
				case 5:
					tag = "mytag5"
				}
				_, err := client.Query(`insert into /test/malert_status_memdb (identifier, severity, lastoccurrence, maintenance, msg, type, agent, firstoccurrence, omni_class, nodealias, alertgroup, node, manager, originalseverity, summary, tags) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) on conflict update firstoccurrence=firstoccurrence, servicenowstate=servicenowstate, emailstatus=emailstatus, tally=tally+1`, "222.129.134.178.1.3.6.1.4.1.2011.5.25.219.2.5.665.25.219.2.5.6"+strconv.Itoa(ii), 5, 1622249715000, 1, "设备:222.129.134.178电源整体功能恢复!", mtype, "huawei mttrapd #"+strconv.Itoa(ii%100), 1622211715023, "300", "222.129.134.178", "huawei_power_func", "222.129.134.178", "MTTrapd Probe", 5, "设备:222.129.134.178电源整体功能恢复!", tag).Do()

				if err != nil {
					wgg.Fail(err)
				} else {
					wgg.Done()
				}
			}(i)

		}
		if err := wgg.Wait(30 * time.Second); err != nil {
			logger.Error(err)
		}
		logger.Debug("concurrent insert malert_status_memdb test, loop ", kk)
	}

	for kk := 0; kk < 10; kk++ {

		for i := kk * DEFAULT_MQL_CONCURRENT_NUM; i < (kk+1)*DEFAULT_MQL_CONCURRENT_NUM; i++ {
			wgg.AddOne()
			go func(ii int) {

				mtype := ii % 10

				tag := "mytag"
				switch mtype {
				case 1:
					tag = "mytag1"
				case 2:
					tag = "mytag2"
				case 3:
					tag = "mytag3"
				case 4:
					tag = "mytag4"
				case 5:
					tag = "mytag5"
				}

				_, err := client.Query(`insert into /test/malert_status_cache (identifier, severity, lastoccurrence, maintenance, msg, type, agent, firstoccurrence, omni_class, nodealias, alertgroup, node, manager, originalseverity, summary, tags) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) on conflict update firstoccurrence=firstoccurrence, servicenowstate=servicenowstate, emailstatus=emailstatus, tally=tally+1`, "222.129.134.178.1.3.6.1.4.1.2011.5.25.219.2.5.665.25.219.2.5.6"+strconv.Itoa(ii), 5, 1622249715000, nil, "设备:222.129.134.178电源整体功能恢复!", mtype, "huawei mttrapd #"+strconv.Itoa(ii%100), 1622211715023, "300", "222.129.134.178", "huawei_power_func", "222.129.134.178", "MTTrapd Probe", 5, "设备:222.129.134.178电源整体功能恢复!", tag).Do()

				if err != nil {
					wgg.Fail(err)
				} else {
					wgg.Done()
				}
			}(i)

		}
		if err := wgg.Wait(30 * time.Second); err != nil {
			logger.Error(err)
		}
		logger.Debug("concurrent insert malert_status_cache test, loop ", kk)
	}

	for kk := 0; kk < 10; kk++ {

		for i := kk * DEFAULT_MQL_CONCURRENT_NUM; i < (kk+1)*DEFAULT_MQL_CONCURRENT_NUM; i++ {
			wgg.AddOne()
			go func(ii int) {

				mtype := (ii + 5) % 10

				tag := "mytag"
				switch mtype {
				case 1:
					tag = "mytag1"
				case 2:
					tag = "mytag2"
				case 3:
					tag = "mytag3"
				case 4:
					tag = "mytag4"
				case 5:
					tag = "mytag5"
				}

				if mtype == 0 {
					_, err = client.Query(`insert into /test/malert_status_cache (identifier, severity, lastoccurrence, maintenance, msg, type, agent, firstoccurrence, omni_class, nodealias, alertgroup, node, manager, originalseverity, summary, tags) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) on conflict update firstoccurrence=firstoccurrence, servicenowstate=servicenowstate, emailstatus=emailstatus, tally=tally+1`, "222.129.134.178.1.3.6.1.4.1.2011.5.25.219.2.5.665.25.219.2.5.6"+strconv.Itoa(ii+100000), 5, 1622249715000, nil, "设备:222.129.134.178电源整体功能恢复!", nil, "huawei mttrapd #"+strconv.Itoa(ii%100), 1622211715023, "300", "222.129.134.178", "huawei_power_func", "222.129.134.178", "MTTrapd Probe", 5, "设备:222.129.134.178电源整体功能恢复!", tag).Do()
				} else {
					_, err = client.Query(`insert into /test/malert_status_cache (identifier, severity, lastoccurrence, maintenance, msg, type, agent, firstoccurrence, omni_class, nodealias, alertgroup, node, manager, originalseverity, summary, tags) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) on conflict update firstoccurrence=firstoccurrence, servicenowstate=servicenowstate, emailstatus=emailstatus, tally=tally+1`, "222.129.134.178.1.3.6.1.4.1.2011.5.25.219.2.5.665.25.219.2.5.6"+strconv.Itoa(ii+100000), 5, 1622249715000, nil, "设备:222.129.134.178电源整体功能恢复!", mtype, "huawei mttrapd #"+strconv.Itoa(ii%100), 1622211715023, "300", "222.129.134.178", "huawei_power_func", "222.129.134.178", "MTTrapd Probe", 5, "设备:222.129.134.178电源整体功能恢复!", tag).Do()
				}
				if err != nil {
					wgg.Fail(err)
				} else {
					wgg.Done()
				}
			}(i)

		}
		if err := wgg.Wait(30 * time.Second); err != nil {
			logger.Error(err)
		}
		logger.Debug("concurrent insert malert_status_cache test, loop ", kk)
	}

	for kk := 0; kk < 10; kk++ {

		for i := kk * DEFAULT_MQL_CONCURRENT_NUM; i < (kk+1)*DEFAULT_MQL_CONCURRENT_NUM; i++ {
			wgg.AddOne()
			go func(ii int) {

				mtype := (ii + 7) % 10

				tag := "mytag"
				switch mtype {
				case 1:
					tag = "mytag1"
				case 2:
					tag = "mytag2"
				case 3:
					tag = "mytag3"
				case 4:
					tag = "mytag4"
				case 5:
					tag = "mytag5"
				}

				if mtype == 0 {
					_, err = client.Query(`insert into /test/malert_status_cache (identifier, severity, lastoccurrence, maintenance, msg, type, agent, firstoccurrence, omni_class, nodealias, alertgroup, node, manager, originalseverity, summary, tags) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) on conflict update firstoccurrence=firstoccurrence, servicenowstate=servicenowstate, emailstatus=emailstatus, tally=tally+1`, "222.129.134.178.1.3.6.1.4.1.2011.5.25.219.2.5.665.25.219.2.5.6"+strconv.Itoa(ii+100000), 5, 1622249715000, nil, "设备:222.129.134.178电源整体功能恢复!", nil, "huawei mttrapd #"+strconv.Itoa(ii%100), 1622211715023, "300", "222.129.134.178", "huawei_power_func", "222.129.134.178", "MTTrapd Probe", 5, "设备:222.129.134.178电源整体功能恢复!", tag).Do()
				} else {
					_, err = client.Query(`insert into /test/malert_status_cache (identifier, severity, lastoccurrence, maintenance, msg, type, agent, firstoccurrence, omni_class, nodealias, alertgroup, node, manager, originalseverity, summary, tags) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) on conflict update firstoccurrence=firstoccurrence, servicenowstate=servicenowstate, emailstatus=emailstatus, tally=tally+1`, "222.129.134.178.1.3.6.1.4.1.2011.5.25.219.2.5.665.25.219.2.5.6"+strconv.Itoa(ii+100000), 5, 1622249715000, nil, "设备:222.129.134.178电源整体功能恢复!", mtype, "huawei mttrapd #"+strconv.Itoa(ii%100), 1622211715023, "300", "222.129.134.178", "huawei_power_func", "222.129.134.178", "MTTrapd Probe", 5, "设备:222.129.134.178电源整体功能恢复!", tag).Do()
				}
				if err != nil {
					wgg.Fail(err)
				} else {
					wgg.Done()
				}
			}(i)

		}
		if err := wgg.Wait(30 * time.Second); err != nil {
			logger.Error(err)
		}
		logger.Debug("concurrent insert malert_status_cache test, loop ", kk)
	}
}