Merge pull request #2370 from luomiao/userdefined-s3endpoint

Support user-defined s3 endpoint

nodeup/pkg/bootstrap/install.go

@@ -17,6 +17,7 @@ limitations under the License.
 package bootstrap
 
 import (
+	"bytes"
 	"fmt"
 	"github.com/golang/glog"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -26,6 +27,7 @@ import (
 	"k8s.io/kops/upup/pkg/fi/nodeup/local"
 	"k8s.io/kops/upup/pkg/fi/nodeup/nodetasks"
 	"k8s.io/kops/util/pkg/vfs"
+	"os"
 	"strings"
 	"time"
 )
@@ -108,6 +110,25 @@ func (i *Installation) buildSystemdJob() *nodetasks.Service {
 	manifest.Set("Unit", "Description", "Run kops bootstrap (nodeup)")
 	manifest.Set("Unit", "Documentation", "https://github.com/kubernetes/kops")
 
+	// Pass in required credentials when using user-defined s3 endpoint
+	if os.Getenv("S3_ENDPOINT") != "" {
+		var buffer bytes.Buffer
+		buffer.WriteString("\"S3_ENDPOINT=")
+		buffer.WriteString(os.Getenv("S3_ENDPOINT"))
+		buffer.WriteString("\" ")
+		buffer.WriteString("\"S3_REGION=")
+		buffer.WriteString(os.Getenv("S3_REGION"))
+		buffer.WriteString("\" ")
+		buffer.WriteString("\"S3_ACCESS_KEY_ID=")
+		buffer.WriteString(os.Getenv("S3_ACCESS_KEY_ID"))
+		buffer.WriteString("\" ")
+		buffer.WriteString("\"S3_SECRET_ACCESS_KEY=")
+		buffer.WriteString(os.Getenv("S3_SECRET_ACCESS_KEY"))
+		buffer.WriteString("\" ")
+
+		manifest.Set("Service", "Environment", buffer.String())
+	}
+
 	manifest.Set("Service", "ExecStart", command)
 	manifest.Set("Service", "Type", "oneshot")
 

nodeup/pkg/model/protokube.go

@@ -17,6 +17,7 @@ limitations under the License.
 package model
 
 import (
+	"bytes"
 	"fmt"
 	"github.com/blang/semver"
 	"github.com/golang/glog"
@@ -26,6 +27,7 @@ import (
 	"k8s.io/kops/pkg/systemd"
 	"k8s.io/kops/upup/pkg/fi"
 	"k8s.io/kops/upup/pkg/fi/nodeup/nodetasks"
+	"os"
 	"strings"
 )
 
@@ -82,6 +84,7 @@ func (b *ProtokubeBuilder) buildSystemdService() (*nodetasks.Service, error) {
 		"--net=host",
 		"--privileged",
 		"--env", "KUBECONFIG=/rootfs/var/lib/kops/kubeconfig",
+		b.ProtokubeEnvironmentVariables(),
 		b.ProtokubeImageName(),
 		"/usr/bin/protokube",
 	}
@@ -218,3 +221,32 @@ func (t *ProtokubeBuilder) ProtokubeFlags(k8sVersion semver.Version) *ProtokubeF
 
 	return f
 }
+
+func (t *ProtokubeBuilder) ProtokubeEnvironmentVariables() string {
+	// Pass in required credentials when using user-defined s3 endpoint
+	if os.Getenv("S3_ENDPOINT") != "" {
+		var buffer bytes.Buffer
+		buffer.WriteString(" ")
+		buffer.WriteString("-e S3_ENDPOINT=")
+		buffer.WriteString("'")
+		buffer.WriteString(os.Getenv("S3_ENDPOINT"))
+		buffer.WriteString("'")
+		buffer.WriteString(" -e S3_REGION=")
+		buffer.WriteString("'")
+		buffer.WriteString(os.Getenv("S3_REGION"))
+		buffer.WriteString("'")
+		buffer.WriteString(" -e S3_ACCESS_KEY_ID=")
+		buffer.WriteString("'")
+		buffer.WriteString(os.Getenv("S3_ACCESS_KEY_ID"))
+		buffer.WriteString("'")
+		buffer.WriteString(" -e S3_SECRET_ACCESS_KEY=")
+		buffer.WriteString("'")
+		buffer.WriteString(os.Getenv("S3_SECRET_ACCESS_KEY"))
+		buffer.WriteString("'")
+		buffer.WriteString(" ")
+
+		return buffer.String()
+	}
+
+	return ""
+}

pkg/model/bootstrapscript.go

@@ -17,10 +17,12 @@ limitations under the License.
 package model
 
 import (
+	"fmt"
 	"k8s.io/kops/pkg/apis/kops"
 	"k8s.io/kops/pkg/apis/nodeup"
 	"k8s.io/kops/pkg/model/resources"
 	"k8s.io/kops/upup/pkg/fi"
+	"os"
 	"text/template"
 )
 
@@ -57,6 +59,18 @@ func (b *BootstrapScript) ResourceNodeUp(ig *kops.InstanceGroup) (*fi.ResourceHo
 
 			return string(data), nil
 		},
+
+		// Pass in extra environment variables for user-defined S3 service
+		"S3Env": func() string {
+			if os.Getenv("S3_ENDPOINT") != "" {
+				return fmt.Sprintf("export S3_ENDPOINT= %s\nexport S3_REGION=%s\nexport S3_ACCESS_KEY_ID=%s\nexport S3_SECRET_ACCESS_KEY=%s\n",
+					os.Getenv("S3_ENDPOINT"),
+					os.Getenv("S3_REGION"),
+					os.Getenv("S3_ACCESS_KEY_ID"),
+					os.Getenv("S3_SECRET_ACCESS_KEY"))
+			}
+			return ""
+		},
 	}
 
 	templateResource, err := NewTemplateResource("nodeup", resources.AWSNodeUpTemplate, functions, nil)

pkg/model/resources/nodeup.go

@@ -38,6 +38,8 @@ set -o pipefail
 NODEUP_URL={{ NodeUpSource }}
 NODEUP_HASH={{ NodeUpSourceHash }}
 
+{{ S3Env }}
+
 function ensure-install-dir() {
   INSTALL_DIR="/var/cache/kubernetes-install"
   # On ContainerOS, we install to /var/lib/toolbox install (because of noexec)

util/pkg/vfs/s3context.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/endpoints"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/ec2"
@@ -34,6 +35,8 @@ type S3Context struct {
 	mutex           sync.Mutex
 	clients         map[string]*s3.S3
 	bucketLocations map[string]string
+
+	getClient func(region string) (*s3.S3, error)
 }
 
 func NewS3Context() *S3Context {
@@ -43,7 +46,7 @@ func NewS3Context() *S3Context {
 	}
 }
 
-func (s *S3Context) getClient(region string) (*s3.S3, error) {
+func (s *S3Context) getDefaultClient(region string) (*s3.S3, error) {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
@@ -61,6 +64,50 @@ func (s *S3Context) getClient(region string) (*s3.S3, error) {
 	return s3Client, nil
 }
 
+func (s *S3Context) getEndpointClient(region string) (*s3.S3, error) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	s3Client := s.clients[region]
+	if s3Client == nil {
+		Endpoint := os.Getenv("S3_ENDPOINT")
+		if Endpoint == "" {
+			return nil, fmt.Errorf("S3_ENDPOINT is required")
+		}
+		AccessKeyID := os.Getenv("S3_ACCESS_KEY_ID")
+		if AccessKeyID == "" {
+			return nil, fmt.Errorf("S3_ACCESS_KEY_ID cannot be empty when S3_ENDPOINT is not empty")
+		}
+		SecretAccessKey := os.Getenv("S3_SECRET_ACCESS_KEY")
+		if SecretAccessKey == "" {
+			return nil, fmt.Errorf("S3_SECRET_ACCESS_KEY cannot be empty when S3_ENDPOINT is not empty")
+		}
+
+		s3Config := &aws.Config{
+			Credentials:      credentials.NewStaticCredentials(AccessKeyID, SecretAccessKey, ""),
+			Endpoint:         aws.String(Endpoint),
+			Region:           aws.String(region),
+			DisableSSL:       aws.Bool(true),
+			S3ForcePathStyle: aws.Bool(true),
+		}
+
+		session := session.New()
+		s3Client = s3.New(session, s3Config)
+		s.clients[region] = s3Client
+	}
+
+	return s3Client, nil
+}
+
+func (s *S3Context) setClientFunc() {
+	Endpoint := os.Getenv("S3_ENDPOINT")
+	if Endpoint == "" {
+		s.getClient = s.getDefaultClient
+	} else {
+		s.getClient = s.getEndpointClient
+	}
+}
+
 func (s *S3Context) getRegionForBucket(bucket string) (string, error) {
 	region := func() string {
 		s.mutex.Lock()

util/pkg/vfs/s3fs.go

@@ -46,6 +46,7 @@ var _ HasHash = &S3Path{}
 func newS3Path(s3Context *S3Context, bucket string, key string) *S3Path {
 	bucket = strings.TrimSuffix(bucket, "/")
 	key = strings.TrimPrefix(key, "/")
+	s3Context.setClientFunc()
 
 	return &S3Path{
 		s3Context: s3Context,