tags.go

package devstatscode

import (
	"database/sql"
	"fmt"
	"strconv"
	"strings"
	"time"
)

// Tags contain list of TSDB tags
type Tags struct {
	Tags []Tag `yaml:"tags"`
}

// Tag contain each TSDB tag data
type Tag struct {
	Name       string               `yaml:"name"`
	SQLFile    string               `yaml:"sql"`
	SeriesName string               `yaml:"series_name"`
	NameTag    string               `yaml:"name_tag"`
	ValueTag   string               `yaml:"value_tag"`
	OtherTags  map[string][2]string `yaml:"other_tags"`
	Limit      int                  `yaml:"limit"`
	Disabled   bool                 `yaml:"disabled"`
}

// ProcessTag - insert given Tag into Postgres TSDB
func ProcessTag(con *sql.DB, ctx *Ctx, tg *Tag, replaces [][]string) {
	// Batch TS points
	var pts TSPoints

	// Skip disabled tags
	if tg.Disabled && !ctx.TestMode {
		return
	}

	// Local or cron mode
	dataPrefix := ctx.DataDir
	if ctx.Local {
		dataPrefix = "./"
	}

	// Per project directory for SQL files
	dir := Metrics
	if ctx.Project != "" {
		dir += ctx.Project + "/"
	}

	// Read SQL file
	bytes, err := ReadFile(ctx, dataPrefix+dir+tg.SQLFile+".sql")
	FatalOnError(err)
	sqlQuery := string(bytes)

	// Handle excluding bots
	bytes, err = ReadFile(ctx, dataPrefix+"util_sql/exclude_bots.sql")
	FatalOnError(err)
	excludeBots := string(bytes)

	// Transform SQL
	limit := tg.Limit
	if limit <= 0 {
		limit = 127
	}
	sqlQuery = strings.Replace(sqlQuery, "{{lim}}", strconv.Itoa(limit), -1)
	sqlQuery = strings.Replace(sqlQuery, "{{exclude_bots}}", excludeBots, -1)

	// Replaces
	for _, replace := range replaces {
		if len(replace) != 2 {
			FatalOnError(fmt.Errorf("replace(s) should have length 2, invalid: %+v", replace))
		}
		sqlQuery = strings.Replace(sqlQuery, replace[0], replace[1], -1)
	}

	// Execute SQL
	rows := QuerySQLWithErr(con, ctx, sqlQuery)
	defer func() { FatalOnError(rows.Close()) }()

	// Drop current tags
	if !ctx.SkipTSDB {
		table := "t" + tg.SeriesName
		if TableExists(con, ctx, table) {
			// ExecSQLWIthErr(con, ctx, "truncate "+table)
			tx, err := con.Begin()
			FatalOnError(err)
			defer func() { _ = tx.Rollback() }()

			_, err = ExecSQLTx(tx, ctx, "set local lock_timeout='500ms'")
			FatalOnError(err)
			_, e := ExecSQLTx(tx, ctx, "truncate "+table)
			if e != nil {
				Printf("truncate failed for %s (warning): %v\n", table, e)
				// The failed truncate aborted the transaction (every further statement
				// would fail with "current transaction is aborted"), so start a new one
				_ = tx.Rollback()
				tx, err = con.Begin()
				FatalOnError(err)
				_, err = ExecSQLTx(tx, ctx, "set local lock_timeout='500ms'")
				FatalOnError(err)
				_, err = ExecSQLTx(tx, ctx, "set local statement_timeout='300s'")
				FatalOnError(err)
				_, e2 := ExecSQLTx(tx, ctx, "delete from "+table)
				if e2 != nil {
					_ = tx.Rollback()
					Printf("delete failed for %s (warning): %v\n", table, e2)
				} else {
					FatalOnError(tx.Commit())
				}
			} else {
				FatalOnError(tx.Commit())
			}
		}
	}
	tm := TimeParseAny("2012-07-01")

	// Columns
	columns, err := rows.Columns()
	FatalOnError(err)
	colIdx := make(map[string]int)
	for i, column := range columns {
		colIdx[column] = i
	}

	// Iterate tag values
	tags := make(map[string]string)
	iVals := make([]interface{}, len(columns))
	for i := range columns {
		iVals[i] = new([]byte)
	}
	got := false
	for rows.Next() {
		got = true
		FatalOnError(rows.Scan(iVals...))
		sVals := []string{}
		for _, iVal := range iVals {
			sVal := ""
			if iVal != nil {
				sVal = string(*iVal.(*[]byte))
			}
			sVals = append(sVals, sVal)
		}
		strVal := sVals[0]
		if tg.NameTag != "" {
			tags[tg.NameTag] = strVal
		}
		if tg.ValueTag != "" {
			tags[tg.ValueTag] = NormalizeName(strVal)
		}
		if tg.OtherTags != nil {
			for tName, tData := range tg.OtherTags {
				tValue := tData[0]
				cIdx, ok := colIdx[tValue]
				if !ok {
					Fatalf("other tag: name: %s: column %s not found", tName, tValue)
				}
				tags[tName] = sVals[cIdx]
				tNorm := strings.ToLower(tData[1])
				if tNorm == "1" || tNorm == "t" || tNorm == "y" {
					tags[tName+"_norm"] = NormalizeName(sVals[cIdx])
				}
			}
		}
		if ctx.Debug > 0 {
			Printf("'%s': %+v\n", tg.SeriesName, tags)
		}
		// Add batch point
		pt := NewTSPoint(ctx, tg.SeriesName, "", tags, nil, tm, false)
		AddTSPoint(ctx, &pts, pt)
		tm = tm.Add(time.Hour)
	}
	FatalOnError(rows.Err())
	if !got {
		Printf("Warning: Tag '%+v' have no values\n", tg)
	}

	// Write the batch
	if !ctx.SkipTSDB {
		WriteTSPoints(ctx, con, &pts, "", []uint8{}, nil)
	} else if ctx.Debug > 0 {
		Printf("Skipping tags series write\n")
	}
}