func TestSQLDropChunkWithLocked(t *testing.T) {
	if testing.Short() {
		t.Skip("skipping integration test")
	}
	if !*useTimescaleDB {
		t.Skip("This test only runs on installs with TimescaleDB")
	}
	withDB(t, *testDatabase, func(db *pgxpool.Pool, t testing.TB) {
		dbJob := testhelpers.PgxPoolWithRole(t, *testDatabase, "prom_maintenance")
		defer dbJob.Close()
		dbSuper, err := pgxpool.Connect(context.Background(), testhelpers.PgConnectURL(*testDatabase, testhelpers.Superuser))
		require.NoError(t, err)
		defer dbSuper.Close()
		oldChunk1 := time.Date(2009, time.November, 11, 0, 0, 0, 0, time.UTC)
		ts := []prompb.TimeSeries{
			{
				Labels: []prompb.Label{
					{Name: pgmodel.MetricNameLabelName, Value: "test"},
					{Name: "name1", Value: "value1"},
				},
				Samples: []prompb.Sample{
					{Timestamp: int64(model.TimeFromUnixNano(oldChunk1.UnixNano()) - 1), Value: 0.1},
					{Timestamp: int64(model.TimeFromUnixNano(time.Now().UnixNano()) - 1), Value: 0.1},
				},
			},
			{
				Labels: []prompb.Label{
					{Name: pgmodel.MetricNameLabelName, Value: "test2"},
					{Name: "name1", Value: "value1"},
				},
				Samples: []prompb.Sample{
					{Timestamp: int64(model.TimeFromUnixNano(time.Now().UnixNano()) - 1), Value: 0.1},
				},
			},
		}
		ingestor, err := ingstr.NewPgxIngestorForTests(pgxconn.NewPgxConn(db), nil)
		require.NoError(t, err)
		defer ingestor.Close()
		_, _, err = ingestor.Ingest(newWriteRequestWithTs(copyMetrics(ts)))
		require.NoError(t, err)
		var tableName string
		var metricId int
		err = db.QueryRow(context.Background(), "SELECT id, table_name FROM _prom_catalog.get_metric_table_name_if_exists('', 'test')").Scan(&metricId, &tableName)
		require.NoError(t, err)
		cnt := 0
		err = db.QueryRow(context.Background(), fmt.Sprintf(`SELECT count(*) FROM show_chunks('prom_data."%s"')`, tableName)).Scan(&cnt)
		require.NoError(t, err)
		require.Equal(t, 2, int(cnt))
		tx, err := db.Begin(context.Background())
		require.NoError(t, err)
		defer func() {
			_ = tx.Rollback(context.Background())
		}()
		_, err = tx.Exec(context.Background(), "SELECT _prom_catalog.lock_metric_for_maintenance($1)", metricId)
		require.NoError(t, err)
		wg := sync.WaitGroup{}
		wg.Add(1)
		maintenanceQuery := "CALL prom_api.execute_maintenance(log_verbose=>true)"
		go func() {
			defer wg.Done()
			_, err := dbJob.Exec(context.Background(), maintenanceQuery)
			require.NoError(t, err)
		}()
		found_wait := false
		for i := 0; i < 5; i++ {
			waiting := false
			err = dbSuper.QueryRow(context.Background(),
				"SELECT wait_event IS NOT NULL FROM pg_stat_activity WHERE query = $1", maintenanceQuery).
				Scan(&waiting)
			if err != pgx.ErrNoRows {
				require.NoError(t, err)
			}
			if waiting {
				found_wait = true
				break
			}
			time.Sleep(time.Second)
		}
		require.True(t, found_wait, "failed to find the exec_maintenance query waiting on lock")
		_, err = tx.Exec(context.Background(), "SELECT _prom_catalog.unlock_metric_for_maintenance($1)", metricId)
		require.NoError(t, err)
		err = tx.Commit(context.Background())
		require.NoError(t, err)
		wg.Wait()
		err = db.QueryRow(context.Background(), fmt.Sprintf(`SELECT count(*) FROM show_chunks('prom_data."%s"')`, tableName)).Scan(&cnt)
		require.NoError(t, err)
		require.Equal(t, 1, int(cnt), "Expected the chunk to be dropped")
	})
}
TestSQLDropChunkWithLocked tests the case where some metrics are locked in the
first loop of execute_data_retention_policy
package end_to_end_tests
import (
	"context"
	"fmt"
	"sync"
	"testing"
	"time"
	"github.com/jackc/pgx/v4"
	"github.com/jackc/pgx/v4/pgxpool"
	_ "github.com/jackc/pgx/v4/stdlib"
	"github.com/prometheus/common/model"
	"github.com/stretchr/testify/require"
	"github.com/timescale/promscale/pkg/internal/testhelpers"
	"github.com/timescale/promscale/pkg/pgmodel/cache"
	ingstr "github.com/timescale/promscale/pkg/pgmodel/ingestor"
	pgmodel "github.com/timescale/promscale/pkg/pgmodel/model"
	"github.com/timescale/promscale/pkg/pgxconn"
	"github.com/timescale/promscale/pkg/prompb"
)
func TestSQLRetentionPeriod(t *testing.T) {
	if testing.Short() {
		t.Skip("skipping integration test")
	}
	withDB(t, *testDatabase, func(dbOwner *pgxpool.Pool, t testing.TB) {
		db := testhelpers.PgxPoolWithRole(t, *testDatabase, "prom_admin")
		defer db.Close()
		ts := []prompb.TimeSeries{
			{
				Labels: []prompb.Label{
					{Name: pgmodel.MetricNameLabelName, Value: "test"},
					{Name: "test", Value: "test"},
				},
				Samples: []prompb.Sample{
					{Timestamp: 1, Value: 0.1},
					{Timestamp: 2, Value: 0.2},
				},
			},
			{
				Labels: []prompb.Label{
					{Name: pgmodel.MetricNameLabelName, Value: "test2"},
					{Name: "test", Value: "test"},
				},
				Samples: []prompb.Sample{
					{Timestamp: 1, Value: 0.1},
					{Timestamp: 2, Value: 0.2},
				},
			},
		}
		ingestor, err := ingstr.NewPgxIngestorForTests(pgxconn.NewPgxConn(db), nil)
		if err != nil {
			t.Fatal(err)
		}
		defer ingestor.Close()
		_, _, err = ingestor.Ingest(newWriteRequestWithTs(copyMetrics(ts)))
		if err != nil {
			t.Fatal(err)
		}
		verifyRetentionPeriod(t, db, "TEST", 90*24*time.Hour)
		_, err = db.Exec(context.Background(), "SELECT prom_api.set_metric_retention_period('prom_data', 'test2', INTERVAL '7 hours')")
		if err != nil {
			t.Error(err)
		}
		verifyRetentionPeriod(t, db, "TEST", 90*24*time.Hour)
		verifyRetentionPeriod(t, db, "test2", 7*time.Hour)
		_, err = db.Exec(context.Background(), "SELECT prom_api.set_default_retention_period(INTERVAL '6 hours')")
		if err != nil {
			t.Error(err)
		}
		verifyRetentionPeriod(t, db, "TEST", 6*time.Hour)
		verifyRetentionPeriod(t, db, "test2", 7*time.Hour)
		_, err = db.Exec(context.Background(), "SELECT prom_api.set_metric_retention_period('TEST', INTERVAL '8 hours')")
		if err != nil {
			t.Error(err)
		}
		_, err = db.Exec(context.Background(), "SELECT prom_api.reset_metric_retention_period('prom_data', 'test2')")
		if err != nil {
			t.Error(err)
		}
		verifyRetentionPeriod(t, db, "test2", 6*time.Hour)
		verifyRetentionPeriod(t, db, "TEST", 8*time.Hour)
		_, err = db.Exec(context.Background(), "SELECT prom_api.reset_metric_retention_period('TEST')")
		if err != nil {
			t.Error(err)
		}
		verifyRetentionPeriod(t, db, "TEST", 6*time.Hour)
		_, err = db.Exec(context.Background(), "SELECT prom_api.set_metric_retention_period('prom_data', 'test_new_metric1', INTERVAL '7 hours')")
		if err != nil {
			t.Error(err)
		}
		verifyRetentionPeriod(t, db, "test_new_metric1", 7*time.Hour)
		_, err = db.Exec(context.Background(), "SELECT prom_api.set_default_retention_period(INTERVAL '2 hours')")
		if err != nil {
			t.Error(err)
		}
		verifyRetentionPeriod(t, db, "test_new_metric1", 7*time.Hour)
		verifyRetentionPeriod(t, db, "test_new_metric2", 2*time.Hour)
	})
}
func verifyRetentionPeriod(t testing.TB, db *pgxpool.Pool, metricName string, expectedDuration time.Duration) {
	var durS int
	var dur time.Duration
	err := db.QueryRow(context.Background(),
		`SELECT EXTRACT(epoch FROM _prom_catalog.get_metric_retention_period('prom_data', $1))`,
		metricName).Scan(&durS)
	if err != nil {
		t.Error(err)
	}
	dur = time.Duration(durS) * time.Second
	if dur != expectedDuration {
		t.Fatalf("Unexpected retention period for table %v: got %v want %v", metricName, dur, expectedDuration)
	}
}
func TestSQLDropChunk(t *testing.T) {
	if testing.Short() {
		t.Skip("skipping integration test")
	}
	if !*useTimescaleDB {
		t.Skip("This test only runs on installs with TimescaleDB")
	}
	withDB(t, *testDatabase, func(db *pgxpool.Pool, t testing.TB) {
		dbJob := testhelpers.PgxPoolWithRole(t, *testDatabase, "prom_maintenance")
		defer dbJob.Close()
		chunkEnds := time.Date(2009, time.November, 11, 0, 0, 0, 0, time.UTC)
		ts := []prompb.TimeSeries{
			{
				Labels: []prompb.Label{
					{Name: pgmodel.MetricNameLabelName, Value: "test"},
					{Name: "name1", Value: "value1"},
				},
				Samples: []prompb.Sample{
					{Timestamp: int64(model.TimeFromUnixNano(chunkEnds.UnixNano()) - 1), Value: 0.1},
					{Timestamp: int64(model.TimeFromUnixNano(time.Now().UnixNano()) - 1), Value: 0.1},
				},
			},
			{
				Labels: []prompb.Label{
					{Name: pgmodel.MetricNameLabelName, Value: "test2"},
					{Name: "name1", Value: "value1"},
				},
				Samples: []prompb.Sample{
					{Timestamp: int64(model.TimeFromUnixNano(time.Now().UnixNano()) - 1), Value: 0.1},
				},
			},
		}
		ingestor, err := ingstr.NewPgxIngestorForTests(pgxconn.NewPgxConn(db), nil)
		if err != nil {
			t.Fatal(err)
		}
		defer ingestor.Close()
		_, _, err = ingestor.Ingest(newWriteRequestWithTs(copyMetrics(ts)))
		if err != nil {
			t.Error(err)
		}
		var tableName string
		err = db.QueryRow(context.Background(), "SELECT table_name FROM _prom_catalog.get_metric_table_name_if_exists('prom_data', 'test')").Scan(&tableName)
		if err != nil {
			t.Fatal(err)
		}
		cnt := 0
		err = db.QueryRow(context.Background(), fmt.Sprintf(`SELECT count(*) FROM show_chunks('prom_data."%s"')`, tableName)).Scan(&cnt)
		if err != nil {
			t.Fatal(err)
		}
		if cnt != 2 {
			t.Errorf("Expected there to be a chunk")
		}
		_, err = dbJob.Exec(context.Background(), "CALL prom_api.execute_maintenance(log_verbose=>true)")
		if err != nil {
			t.Fatal(err)
		}
		err = db.QueryRow(context.Background(), fmt.Sprintf(`SELECT count(*) FROM show_chunks('prom_data."%s"')`, tableName)).Scan(&cnt)
		if err != nil {
			t.Fatal(err)
		}
		if cnt != 1 {
			t.Errorf("Expected the chunk to be dropped")
		}
		_, err = dbJob.Exec(context.Background(), "CALL prom_api.execute_maintenance()")
		if err != nil {
			t.Fatal(err)
		}
		err = db.QueryRow(context.Background(), "SELECT count(*) FROM show_chunks('prom_data.test2')").Scan(&cnt)
		if err != nil {
			t.Fatal(err)
		}
		if cnt != 1 {
			t.Errorf("Expected the chunk to be dropped")
		}
	})
}