Skip to content

Commit

Permalink
feat(controller): Store artifact repostiory in workflow status. Fixes a…
Browse files Browse the repository at this point in the history 
…rgoproj#6255

Signed-off-by: Alex Collins <[email protected]>
  • Loading branch information
@alexec
alexec committed Jul 7, 2021
1 parent c9246d3 commit 66582b1
Show file tree
Hide file tree
Showing 15 changed files with 2,535 additions and 926 deletions.
146 changes: 1 addition & 145 deletions config/config.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,15 +3,13 @@ package config
import (
"fmt"
"math"
"path"
"time"

apiv1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"

wfv1 "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
"github.com/argoproj/argo-workflows/v3/workflow/common"
)

var EmptyConfigFunc = func() interface{} { return &Config{} }
Expand Down Expand Up @@ -60,7 +58,7 @@ type Config struct {
KubeletInsecure bool `json:"kubeletInsecure,omitempty"`

// ArtifactRepository contains the default location of an artifact repository for container artifacts
ArtifactRepository ArtifactRepository `json:"artifactRepository,omitempty"`
ArtifactRepository wfv1.ArtifactRepository `json:"artifactRepository,omitempty"`

// Namespace is a label selector filter to limit the controller's watch to a specific namespace
// DEPRECATED: support will be remove in a future release
Expand Down Expand Up @@ -178,61 +176,6 @@ type KubeConfig struct {
MountPath string `json:"mountPath,omitempty"`
}

// ArtifactRepository represents an artifact repository in which a controller will store its artifacts
type ArtifactRepository struct {
// ArchiveLogs enables log archiving
ArchiveLogs *bool `json:"archiveLogs,omitempty"`
// S3 stores artifact in a S3-compliant object store
S3 *S3ArtifactRepository `json:"s3,omitempty"`
// Artifactory stores artifacts to JFrog Artifactory
Artifactory *ArtifactoryArtifactRepository `json:"artifactory,omitempty"`
// HDFS stores artifacts in HDFS
HDFS *HDFSArtifactRepository `json:"hdfs,omitempty"`
// OSS stores artifact in a OSS-compliant object store
OSS *OSSArtifactRepository `json:"oss,omitempty"`
// GCS stores artifact in a GCS object store
GCS *GCSArtifactRepository `json:"gcs,omitempty"`
}

func (a *ArtifactRepository) IsArchiveLogs() bool {
return a != nil && a.ArchiveLogs != nil && *a.ArchiveLogs
}

type ArtifactRepositoryType interface {
IntoArtifactLocation(l *wfv1.ArtifactLocation)
}

func (a *ArtifactRepository) Get() ArtifactRepositoryType {
if a == nil {
return nil
} else if a.Artifactory != nil {
return a.Artifactory
} else if a.GCS != nil {
return a.GCS
} else if a.HDFS != nil {
return a.HDFS
} else if a.OSS != nil {
return a.OSS
} else if a.S3 != nil {
return a.S3
}
return nil
}

// ToArtifactLocation returns the artifact location set with default template key:
// key = `{{workflow.name}}/{{pod.name}}`
func (a *ArtifactRepository) ToArtifactLocation() *wfv1.ArtifactLocation {
if a == nil {
return nil
}
l := &wfv1.ArtifactLocation{ArchiveLogs: a.ArchiveLogs}
v := a.Get()
if v != nil {
v.IntoArtifactLocation(l)
}
return l
}

type PersistConfig struct {
NodeStatusOffload bool `json:"nodeStatusOffLoad,omitempty"`
// Archive workflows to persistence.
Expand Down Expand Up @@ -295,93 +238,6 @@ type MySQLConfig struct {
Options map[string]string `json:"options,omitempty"`
}

// S3ArtifactRepository defines the controller configuration for an S3 artifact repository
type S3ArtifactRepository struct {
wfv1.S3Bucket `json:",inline"`

// KeyFormat is defines the format of how to store keys. Can reference workflow variables
KeyFormat string `json:"keyFormat,omitempty"`

// KeyPrefix is prefix used as part of the bucket key in which the controller will store artifacts.
// DEPRECATED. Use KeyFormat instead
KeyPrefix string `json:"keyPrefix,omitempty"`
}

func (r *S3ArtifactRepository) IntoArtifactLocation(l *wfv1.ArtifactLocation) {
k := r.KeyFormat
if k == "" {
k = path.Join(r.KeyPrefix, common.DefaultArchivePattern)
}
l.S3 = &wfv1.S3Artifact{S3Bucket: r.S3Bucket, Key: k}
}

// OSSArtifactRepository defines the controller configuration for an OSS artifact repository
type OSSArtifactRepository struct {
wfv1.OSSBucket `json:",inline"`

// KeyFormat is defines the format of how to store keys. Can reference workflow variables
KeyFormat string `json:"keyFormat,omitempty"`
}

func (r *OSSArtifactRepository) IntoArtifactLocation(l *wfv1.ArtifactLocation) {
k := r.KeyFormat
if k == "" {
k = common.DefaultArchivePattern
}
l.OSS = &wfv1.OSSArtifact{OSSBucket: r.OSSBucket, Key: k}
}

// GCSArtifactRepository defines the controller configuration for a GCS artifact repository
type GCSArtifactRepository struct {
wfv1.GCSBucket `json:",inline"`

// KeyFormat is defines the format of how to store keys. Can reference workflow variables
KeyFormat string `json:"keyFormat,omitempty"`
}

func (r *GCSArtifactRepository) IntoArtifactLocation(l *wfv1.ArtifactLocation) {
k := r.KeyFormat
if k == "" {
k = common.DefaultArchivePattern
}
l.GCS = &wfv1.GCSArtifact{GCSBucket: r.GCSBucket, Key: k}
}

// ArtifactoryArtifactRepository defines the controller configuration for an artifactory artifact repository
type ArtifactoryArtifactRepository struct {
wfv1.ArtifactoryAuth `json:",inline"`
// RepoURL is the url for artifactory repo.
RepoURL string `json:"repoURL,omitempty"`
}

func (r *ArtifactoryArtifactRepository) IntoArtifactLocation(l *wfv1.ArtifactLocation) {
u := ""
if r.RepoURL != "" {
u = r.RepoURL + "/"
}
u = fmt.Sprintf("%s%s", u, common.DefaultArchivePattern)
l.Artifactory = &wfv1.ArtifactoryArtifact{ArtifactoryAuth: r.ArtifactoryAuth, URL: u}
}

// HDFSArtifactRepository defines the controller configuration for an HDFS artifact repository
type HDFSArtifactRepository struct {
wfv1.HDFSConfig `json:",inline"`

// PathFormat is defines the format of path to store a file. Can reference workflow variables
PathFormat string `json:"pathFormat,omitempty"`

// Force copies a file forcibly even if it exists (default: false)
Force bool `json:"force,omitempty"`
}

func (r *HDFSArtifactRepository) IntoArtifactLocation(l *wfv1.ArtifactLocation) {
p := r.PathFormat
if p == "" {
p = common.DefaultArchivePattern
}
l.HDFS = &wfv1.HDFSArtifact{HDFSConfig: r.HDFSConfig, Path: p, Force: r.Force}
}

// MetricsConfig defines a config for a metrics server
type MetricsConfig struct {
// Enabled controls metric emission. Default is true, set "enabled: false" to turn off
Expand Down
61 changes: 0 additions & 61 deletions config/config_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,69 +6,8 @@ import (
"github.com/stretchr/testify/assert"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/utils/pointer"
)

func TestArtifactRepository(t *testing.T) {
t.Run("Nil", func(t *testing.T) {
var r *ArtifactRepository
assert.Nil(t, r.Get())
l := r.ToArtifactLocation()
assert.Nil(t, l)
})
t.Run("ArchiveLogs", func(t *testing.T) {
r := &ArtifactRepository{Artifactory: &ArtifactoryArtifactRepository{}, ArchiveLogs: pointer.BoolPtr(true)}
l := r.ToArtifactLocation()
assert.Equal(t, pointer.BoolPtr(true), l.ArchiveLogs)
})
t.Run("Artifactory", func(t *testing.T) {
r := &ArtifactRepository{Artifactory: &ArtifactoryArtifactRepository{RepoURL: "http://my-repo"}}
assert.IsType(t, &ArtifactoryArtifactRepository{}, r.Get())
l := r.ToArtifactLocation()
if assert.NotNil(t, l.Artifactory) {
assert.Equal(t, "http://my-repo/{{workflow.name}}/{{pod.name}}", l.Artifactory.URL)
}
})
t.Run("GCS", func(t *testing.T) {
r := &ArtifactRepository{GCS: &GCSArtifactRepository{}}
assert.IsType(t, &GCSArtifactRepository{}, r.Get())
l := r.ToArtifactLocation()
if assert.NotNil(t, l.GCS) {
assert.Equal(t, "{{workflow.name}}/{{pod.name}}", l.GCS.Key)
}
})
t.Run("HDFS", func(t *testing.T) {
r := &ArtifactRepository{HDFS: &HDFSArtifactRepository{}}
assert.IsType(t, &HDFSArtifactRepository{}, r.Get())
l := r.ToArtifactLocation()
if assert.NotNil(t, l.HDFS) {
assert.Equal(t, "{{workflow.name}}/{{pod.name}}", l.HDFS.Path)
}
})
t.Run("OSS", func(t *testing.T) {
r := &ArtifactRepository{OSS: &OSSArtifactRepository{}}
assert.IsType(t, &OSSArtifactRepository{}, r.Get())
l := r.ToArtifactLocation()
if assert.NotNil(t, l.OSS) {
assert.Equal(t, "{{workflow.name}}/{{pod.name}}", l.OSS.Key)
}
})
t.Run("S3", func(t *testing.T) {
r := &ArtifactRepository{S3: &S3ArtifactRepository{KeyPrefix: "my-key-prefix"}}
assert.IsType(t, &S3ArtifactRepository{}, r.Get())
l := r.ToArtifactLocation()
if assert.NotNil(t, l.S3) {
assert.Equal(t, "my-key-prefix/{{workflow.name}}/{{pod.name}}", l.S3.Key)
}
})
}

func TestArtifactRepository_IsArchiveLogs(t *testing.T) {
assert.False(t, (&ArtifactRepository{}).IsArchiveLogs())
assert.False(t, (&ArtifactRepository{ArchiveLogs: pointer.BoolPtr(false)}).IsArchiveLogs())
assert.True(t, (&ArtifactRepository{ArchiveLogs: pointer.BoolPtr(true)}).IsArchiveLogs())
}

func TestDatabaseConfig(t *testing.T) {
assert.Equal(t, "my-host", DatabaseConfig{Host: "my-host"}.GetHostname())
assert.Equal(t, "my-host:1234", DatabaseConfig{Host: "my-host", Port: 1234}.GetHostname())
Expand Down
Loading

0 comments on commit 66582b1

Please sign in to comment.