cidataloader

README

ci-data-loader

A derivative of https://github.com/openshift/ci-search-functions

The cloud functions in this repository are used to load known prow job result artifacts to the specified big query project.

The functions operate on origin-ci-test and must therefore be deployed in the openshift-gce-devel project. The service account job-run-big-query-writer@openshift-gce-devel.iam.gserviceaccount.com was created ahead of time and given storage viewer access on the origin-ci-test bucket. The BigQuery project specified in the deployment env-vars will need to add the service account and grant BigQuery Data Editor role to the SA for the specified dataset.

During first deployment the function should not be accessible to external viewers.

Documentation

Index

• Constants
• func LoadJobRunData(ctx context.Context, e GCSEvent) error
• func LoadJobRunDataTest(ctx context.Context, e GCSEvent) error
• func UnMarshalJSON(jsonb []byte, result interface{}) error
• type BigQueryDataItem
  • func (i *BigQueryDataItem) Save() (row map[string]bigquery.Value, insertID string, err error)
• type BigQueryLoader
  • func (b *BigQueryLoader) FindExistingData(ctx context.Context, partitionTime time.Time, ...) (bool, error)
  • func (b *BigQueryLoader) GetMetaData(ctx context.Context, di DataInstance) (bigquery.TableMetadata, error)
  • func (b *BigQueryLoader) Init(ctx context.Context) error
  • func (b *BigQueryLoader) LoadDataItems(ctx context.Context, dataInstance DataInstance) ([]interface{}, error)
  • func (b *BigQueryLoader) ValidateTable(ctx context.Context, dataInstance DataInstance) error
• type BigQueryTableCache
• type ClientError
  • func (c *ClientError) Error() string
• type ClientsCache
• type DataFile
• type DataInstance
• type DataLoader
• type DataType
• type DataUploader
• type GCSEvent
• type JobRunDataEvent
• type OutputMetric
• type PrometheusData
• type PrometheusLabels
  • func (l PrometheusLabels) MarshalJSON() ([]byte, error)
  • func (l *PrometheusLabels) UnmarshalJSON(data []byte) error
• type PrometheusMetric
• type PrometheusResult
• type PrometheusValue
  • func (l *PrometheusValue) UnmarshalJSON(data []byte) error
• type SimpleUploader

Constants

const (
	AutoDataLoaderSuffix  = "autodl.json"
	DataSetEnv            = "DATASET_ID"
	ProjectIdEnv          = "PROJECT_ID"
	PRJobsEnabledEnv      = "PR_JOBS_ENABLED"      // local testing only
	GCSCredentialsFileEnv = "GCS_CREDENTIALS_FILE" // local testing only
)

const DataPartitioningField = "PartitionTime"

const JobRunNameField = "JobRunName"

const SourceNameField = "Source"

Functions

func LoadJobRunData

func LoadJobRunData(ctx context.Context, e GCSEvent) error

func LoadJobRunDataTest

func LoadJobRunDataTest(ctx context.Context, e GCSEvent) error

func UnMarshalJSON

func UnMarshalJSON(jsonb []byte, result interface{}) error

Types

type BigQueryDataItem

type BigQueryDataItem struct {
	Instance *DataInstance
	Row      map[string]string
	InsertID string
}

func (*BigQueryDataItem) Save

func (i *BigQueryDataItem) Save() (row map[string]bigquery.Value, insertID string, err error)

Save implements the ValueSaver interface.

type BigQueryLoader

type BigQueryLoader struct {
	ProjectID string
	DataSetID string
	Client    *bigquery.Client
	DryRun    bool
	// contains filtered or unexported fields
}

func (*BigQueryLoader) FindExistingData

func (b *BigQueryLoader) FindExistingData(ctx context.Context, partitionTime time.Time, partitionColumn, tableName, jobRunName, source string) (bool, error)

func (*BigQueryLoader) GetMetaData

func (b *BigQueryLoader) GetMetaData(ctx context.Context, di DataInstance) (bigquery.TableMetadata, error)

func (*BigQueryLoader) Init

func (b *BigQueryLoader) Init(ctx context.Context) error

func (*BigQueryLoader) LoadDataItems

func (b *BigQueryLoader) LoadDataItems(ctx context.Context, dataInstance DataInstance) ([]interface{}, error)

func (*BigQueryLoader) ValidateTable

func (b *BigQueryLoader) ValidateTable(ctx context.Context, dataInstance DataInstance) error

type BigQueryTableCache

type BigQueryTableCache struct {
	// contains filtered or unexported fields
}

type ClientError

type ClientError struct {
	Err  error
	Call string
}

func (*ClientError) Error

func (c *ClientError) Error() string

type ClientsCache

type ClientsCache struct {
	// contains filtered or unexported fields
}

type DataFile

type DataFile struct {
	TableName       string              `json:"table_name"`
	Schema          map[string]DataType `json:"schema"`
	SchemaMapping   map[string]string   `json:"schema_mapping"`
	Rows            []map[string]string `json:"rows"`
	ExpirationDays  int                 `json:"expiration_days"`
	PartitionColumn string              `json:"partition_column"`
	PartitionType   string              `json:"partition_type"`
	ChunkSize       int                 `json:"chunk_size"`
}

type DataInstance

type DataInstance struct {
	Source       string
	CreationTime time.Time
	JobRunName   string
	DataFile     *DataFile
}

type DataLoader

type DataLoader interface {
	ValidateTable(ctx context.Context, dataInstance DataInstance) error
	FindExistingData(ctx context.Context, partitionTime time.Time, partitionColumn, tableName, jobRunName, source string) (bool, error)
	LoadDataItems(ctx context.Context, dataInstance DataInstance) ([]interface{}, error)
}

type DataType

type DataType = string

const (
	DataTypeFloat64 DataType = "float64"
	DataTypeString  DataType = "string"
	DataTypeInteger DataType = "int64"
	// RFC3339  based value "2006-01-02T15:04:05Z07:00
	DataTypeTimestamp DataType = "timestamp"
	DataTypeJSON      DataType = "json"
	DataTypeBool      DataType = "bool"
)

type DataUploader

type DataUploader struct {
	// contains filtered or unexported fields
}

type GCSEvent

type GCSEvent struct {
	Kind string `json:"kind"`
	ID   string `json:"id"`
	// SelfLink                string                 `json:"selfLink"`
	Name   string `json:"name"`
	Bucket string `json:"bucket"`
	// Generation              string                 `json:"generation"`
	// Metageneration          string                 `json:"metageneration"`
	ContentType string    `json:"contentType"`
	TimeCreated time.Time `json:"timeCreated"`
	Updated     time.Time `json:"updated"`
	// TemporaryHold           bool                   `json:"temporaryHold"`
	// EventBasedHold          bool                   `json:"eventBasedHold"`
	// RetentionExpirationTime time.Time              `json:"retentionExpirationTime"`
	// StorageClass            string                 `json:"storageClass"`
	// TimeStorageClassUpdated time.Time              `json:"timeStorageClassUpdated"`
	Size      string `json:"size"`
	MD5Hash   string `json:"md5Hash"`
	MediaLink string `json:"mediaLink"`
	// ContentEncoding         string                 `json:"contentEncoding"`
	// ContentDisposition      string                 `json:"contentDisposition"`
	// CacheControl            string                 `json:"cacheControl"`
	Metadata map[string]interface{} `json:"metadata"`
}

type JobRunDataEvent

type JobRunDataEvent struct {
	Job      string
	BuildID  string
	Filename string
	GCSEvent *GCSEvent
}

type OutputMetric

type OutputMetric struct {
	Timestamp int64  `json:"timestamp"`
	Value     string `json:"value"`
}

type PrometheusData

type PrometheusData struct {
	ResultType string             `json:"resultType"`
	Result     []PrometheusMetric `json:"result"`
}

type PrometheusLabels

type PrometheusLabels map[string]string

PrometheusLabels avoids deserialization allocations

func (PrometheusLabels) MarshalJSON

func (l PrometheusLabels) MarshalJSON() ([]byte, error)

func (*PrometheusLabels) UnmarshalJSON

func (l *PrometheusLabels) UnmarshalJSON(data []byte) error

type PrometheusMetric

type PrometheusMetric struct {
	Metric PrometheusLabels `json:"metric"`
	Value  PrometheusValue  `json:"value"`
}

type PrometheusResult

type PrometheusResult struct {
	Status string         `json:"status"`
	Data   PrometheusData `json:"data"`
}

type PrometheusValue

type PrometheusValue struct {
	Timestamp int64
	Value     string
}

func (*PrometheusValue) UnmarshalJSON

func (l *PrometheusValue) UnmarshalJSON(data []byte) error

type SimpleUploader

type SimpleUploader interface {
	// contains filtered or unexported methods
}