(promTag, table, op, value, originFilter string, e *CHEngine)
| 1280 | } |
| 1281 | |
| 1282 | func GetRemoteReadFilter(promTag, table, op, value, originFilter string, e *CHEngine) (string, error) { |
| 1283 | filter := "" |
| 1284 | sql := "" |
| 1285 | isAppLabel := false |
| 1286 | nameNoPrefix := strings.TrimPrefix(promTag, "tag.") |
| 1287 | metricID, ok := trans_prometheus.ORGPrometheus[e.ORGID].MetricNameToID[table] |
| 1288 | if !ok { |
| 1289 | errorMessage := fmt.Sprintf("%s not found", table) |
| 1290 | return filter, common.NewError(common.RESOURCE_NOT_FOUND, errorMessage) |
| 1291 | } |
| 1292 | labelNameID, ok := trans_prometheus.ORGPrometheus[e.ORGID].LabelNameToID[nameNoPrefix] |
| 1293 | if !ok { |
| 1294 | if value == "''" { |
| 1295 | filter = fmt.Sprintf("1%s1", op) |
| 1296 | } else { |
| 1297 | filter = "1!=1" |
| 1298 | } |
| 1299 | debugMessage := fmt.Sprintf("%s not found", nameNoPrefix) |
| 1300 | log.Debug(debugMessage) |
| 1301 | return filter, nil |
| 1302 | } |
| 1303 | prometheusSubqueryCache := GetPrometheusSubqueryCache() |
| 1304 | // Determine whether the tag is app_label or target_label |
| 1305 | if appLabels, ok := trans_prometheus.ORGPrometheus[e.ORGID].MetricAppLabelLayout[table]; ok { |
| 1306 | for _, appLabel := range appLabels { |
| 1307 | if appLabel.AppLabelName == nameNoPrefix { |
| 1308 | isAppLabel = true |
| 1309 | entryKey := common.EntryKey{ORGID: e.ORGID, Filter: originFilter} |
| 1310 | cacheFilter, ok := prometheusSubqueryCache.Get(entryKey) |
| 1311 | if ok { |
| 1312 | filter = cacheFilter.Filter |
| 1313 | timeout := cacheFilter.Time |
| 1314 | if time.Since(timeout) < time.Duration(config.Cfg.PrometheusIdSubqueryLruTimeout) { |
| 1315 | return filter, nil |
| 1316 | } |
| 1317 | } |
| 1318 | if value == "''" { |
| 1319 | filter = fmt.Sprintf("app_label_value_id_%d %s 0", appLabel.AppLabelColumnIndex, op) |
| 1320 | entryValue := common.EntryValue{Time: time.Now(), Filter: filter} |
| 1321 | entryKey := common.EntryKey{ORGID: e.ORGID, Filter: originFilter} |
| 1322 | prometheusSubqueryCache.Add(entryKey, entryValue) |
| 1323 | return filter, nil |
| 1324 | } |
| 1325 | |
| 1326 | // lru timeout |
| 1327 | if strings.Contains(op, "match") { |
| 1328 | sql = fmt.Sprintf("SELECT label_value_id FROM flow_tag.app_label_live_view WHERE label_name_id=%d and %s(label_value,%s) GROUP BY label_value_id", labelNameID, op, value) |
| 1329 | } else { |
| 1330 | sql = fmt.Sprintf("SELECT label_value_id FROM flow_tag.app_label_live_view WHERE label_name_id=%d and label_value %s %s GROUP BY label_value_id", labelNameID, op, value) |
| 1331 | } |
| 1332 | chClient := client.Client{ |
| 1333 | Host: config.Cfg.Clickhouse.Host, |
| 1334 | Port: config.Cfg.Clickhouse.Port, |
| 1335 | UserName: config.Cfg.Clickhouse.User, |
| 1336 | Password: config.Cfg.Clickhouse.Password, |
| 1337 | DB: "flow_tag", |
| 1338 | } |
| 1339 | appLabelRst, err := chClient.DoQuery(&client.QueryParams{Sql: sql, ORGID: e.ORGID}) |
no test coverage detected