func (c *Client) Ingest(r *prompb.WriteRequest) (uint64, uint64, error) {
	return c.ingestor.Ingest(r)
}
Ingest writes the timeseries object into the DB
package pgclient
import (
	"context"
	"fmt"
	"net/url"
	"github.com/jackc/pgx/v4"
	"github.com/jackc/pgx/v4/pgxpool"
	"github.com/timescale/promscale/pkg/ha"
	"github.com/timescale/promscale/pkg/log"
	"github.com/timescale/promscale/pkg/pgmodel/cache"
	"github.com/timescale/promscale/pkg/pgmodel/health"
	"github.com/timescale/promscale/pkg/pgmodel/ingestor"
	"github.com/timescale/promscale/pkg/pgmodel/lreader"
	"github.com/timescale/promscale/pkg/pgmodel/querier"
	"github.com/timescale/promscale/pkg/pgxconn"
	"github.com/timescale/promscale/pkg/prompb"
	"github.com/timescale/promscale/pkg/promql"
	"github.com/timescale/promscale/pkg/query"
	"github.com/timescale/promscale/pkg/tenancy"
	"go.opentelemetry.io/collector/model/pdata"
)
type Client struct {
	Connection        pgxconn.PgxConn
	QuerierConnection pgxconn.PgxConn
	ingestor          *ingestor.DBIngestor
	querier           querier.Querier
	healthCheck       health.HealthCheckerFn
	queryable         promql.Queryable
	ConnectionStr     string
	metricCache       cache.MetricCache
	labelsCache       cache.LabelsCache
	seriesCache       cache.SeriesCache
	closePool         bool
	sigClose          chan struct{}
	haService         *ha.Service
}
type LockFunc = func(ctx context.Context, conn *pgx.Conn) error
func NewClient(cfg *Config, mt tenancy.Authorizer, schemaLocker LockFunc, readOnly bool) (*Client, error) {
	pgConfig, numCopiers, err := getPgConfig(cfg)
	if err != nil {
		return nil, err
	}
	pgConfig.AfterConnect = schemaLocker
	connectionPool, err := pgxpool.ConnectConfig(context.Background(), pgConfig)
	if err != nil {
		log.Error("msg", "err creating connection pool for new client", "err", err.Error())
		return nil, err
	}
	client, err := NewClientWithPool(cfg, numCopiers, connectionPool, mt, readOnly)
	if err != nil {
		return client, err
	}
	client.closePool = true
	return client, err
}
func getPgConfig(cfg *Config) (*pgxpool.Config, int, error) {
	minConnections, maxConnections, numCopiers, err := cfg.GetNumConnections()
	if err != nil {
		log.Error("msg", "configuring number of connections", "err", err.Error())
		return nil, numCopiers, err
	}
	connectionStr := cfg.GetConnectionStr()
	pgConfig, err := pgxpool.ParseConfig(connectionStr)
	if err != nil {
		log.Error("msg", "configuring connection", "err", err.Error())
		return nil, numCopiers, err
	}
	pgConfig.MinConns = int32(minConnections)
	pgConfig.MaxConns = int32(maxConnections)
	var statementCacheLog string
	if cfg.EnableStatementsCache {
		statementCacheCapacity := 512
		pgConfig.AfterRelease = observeStatementCacheState
		statementCacheEnabled.Set(1)
		statementCacheCap.Set(float64(statementCacheCapacity))
		statementCacheLog = fmt.Sprintf("%d statements", statementCacheCapacity)
	} else {
		log.Info("msg", "Statements cached disabled, using simple protocol for database connections.")
		pgConfig.ConnConfig.PreferSimpleProtocol = true
		statementCacheEnabled.Set(0)
		statementCacheCap.Set(0)
		statementCacheLog = "disabled"
	}
	log.Info("msg", getRedactedConnStr(connectionStr),
		"numCopiers", numCopiers,
		"pool_max_conns", maxConnections,
		"pool_min_conns", minConnections,
		"statement_cache", statementCacheLog,
	)
	return pgConfig, numCopiers, nil
}
func getRedactedConnStr(s string) string {
	connURL, err := url.Parse(s)
	if err != nil {
		return "****"
	}
	if _, pwSet := connURL.User.Password(); pwSet {
		connURL.User = url.UserPassword(connURL.User.Username(), "****")
	}
	return connURL.String()
}
func NewClientWithPool(cfg *Config, numCopiers int, connPool *pgxpool.Pool, mt tenancy.Authorizer, readOnly bool) (*Client, error) {
	dbConn := pgxconn.NewPgxConn(connPool)
	sigClose := make(chan struct{})
	metricsCache := cache.NewMetricCache(cfg.CacheConfig)
	labelsCache := cache.NewLabelsCache(cfg.CacheConfig)
	seriesCache := cache.NewSeriesCache(cfg.CacheConfig, sigClose)
	c := ingestor.Cfg{
		NumCopiers:             numCopiers,
		IgnoreCompressedChunks: cfg.IgnoreCompressedChunks,
	}
	var (
		dbIngestor          *ingestor.DBIngestor
		exemplarKeyPosCache = cache.NewExemplarLabelsPosCache(cfg.CacheConfig)
	)
	if !readOnly {
		var err error
		dbIngestor, err = ingestor.NewPgxIngestor(dbConn, metricsCache, seriesCache, exemplarKeyPosCache, &c)
		if err != nil {
			log.Error("msg", "err starting the ingestor", "err", err)
			return nil, err
		}
	}
	labelsReader := lreader.NewLabelsReader(dbConn, labelsCache)
	dbQuerierConn := pgxconn.NewQueryLoggingPgxConn(connPool)
	dbQuerier := querier.NewQuerier(dbQuerierConn, metricsCache, labelsReader, exemplarKeyPosCache, mt.ReadAuthorizer())
	queryable := query.NewQueryable(dbQuerier, labelsReader)
	healthChecker := health.NewHealthChecker(dbConn)
	client := &Client{
		Connection:        dbConn,
		QuerierConnection: dbQuerierConn,
		ingestor:          dbIngestor,
		querier:           dbQuerier,
		healthCheck:       healthChecker,
		queryable:         queryable,
		metricCache:       metricsCache,
		labelsCache:       labelsCache,
		seriesCache:       seriesCache,
		sigClose:          sigClose,
	}
	InitClientMetrics(client)
	return client, nil
}
func (c *Client) Close() {
	log.Info("msg", "Shutting down Client")
	if c.ingestor != nil {
		c.ingestor.Close()
	}
	close(c.sigClose)
	if c.closePool {
		c.Connection.Close()
	}
	if c.haService != nil {
		c.haService.Close()
	}
}
func (c *Client) Ingestor() *ingestor.DBIngestor {
	return c.ingestor
}