Merge remote-tracking branch 'google/master'

cmd/cloud_sql_proxy/cloud_sql_proxy.go

@@ -20,16 +20,13 @@
 package main
 
 import (
-	"bytes"
-	"encoding/json"
 	"errors"
 	"flag"
 	"fmt"
 	"io/ioutil"
 	"log"
 	"net/http"
 	"os"
-	"os/exec"
 	"path/filepath"
 	"strings"
 	"sync"
@@ -38,7 +35,9 @@ import (
 	"github.com/GoogleCloudPlatform/cloudsql-proxy/logging"
 	"github.com/GoogleCloudPlatform/cloudsql-proxy/proxy/certs"
 	"github.com/GoogleCloudPlatform/cloudsql-proxy/proxy/fuse"
+	"github.com/GoogleCloudPlatform/cloudsql-proxy/proxy/limits"
 	"github.com/GoogleCloudPlatform/cloudsql-proxy/proxy/proxy"
+	"github.com/GoogleCloudPlatform/cloudsql-proxy/proxy/util"
 
 	"cloud.google.com/go/compute/metadata"
 	"golang.org/x/net/context"
@@ -76,17 +75,21 @@ can be removed automatically by this program.`)
 
 	// Settings for limits
 	maxConnections = flag.Uint64("max_connections", 0, `If provided, the maximum number of connections to establish before refusing new connections. Defaults to 0 (no limit)`)
+	fdRlimit       = flag.Uint64("fd_rlimit", limits.ExpectedFDs, `Sets the rlimit on the number of open file descriptors for the proxy to the provided value. If set to zero, disables attempts to set the rlimit. Defaults to a value which can support 4K connections to one instance`)
 
 	// Settings for authentication.
 	token     = flag.String("token", "", "When set, the proxy uses this Bearer token for authorization.")
 	tokenFile = flag.String("credential_file", "", `If provided, this json file will be used to retrieve Service Account credentials.
 You may set the GOOGLE_APPLICATION_CREDENTIALS environment variable for the same effect.`)
+	ipAddressTypes = flag.String("ip_address_types", "PRIMARY", "Default to be 'PRIMARY'. Options: a list of strings separated by ',', e.g. 'PRIMARY, PRIVATE' ")
+
+	// Setting to choose what API to connect to
+	host = flag.String("host", "https://www.googleapis.com/sql/v1beta4/", "When set, the proxy uses this host as the base API path.")
 )
 
 const (
 	minimumRefreshCfgThrottle = time.Second
 
-	host = "https://www.googleapis.com/sql/v1beta4/"
 	port = 3307
 )
 
@@ -167,7 +170,6 @@ Connection:
     When using Unix sockets (the default for systems which support them), the
     Proxy places the sockets in the directory specified by the -dir parameter.
 
-
 Automatic instance discovery:
    If the Google Cloud SQL is installed on the local machine and no instance
    connection flags are specified, the proxy connects to all instances in the
@@ -270,7 +272,17 @@ func authenticatedClient(ctx context.Context) (*http.Client, error) {
 		return oauth2.NewClient(ctx, src), nil
 	}
 
-	return goauth.DefaultClient(ctx, proxy.SQLScope)
+	// If flags don't specify an auth source, try either gcloud or application default
+	// credentials.
+	src, err := util.GcloudTokenSource(ctx)
+	if err != nil {
+		src, err = goauth.DefaultTokenSource(ctx, proxy.SQLScope)
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	return oauth2.NewClient(ctx, src), nil
 }
 
 func stringList(s string) []string {
@@ -327,34 +339,15 @@ func listInstances(ctx context.Context, cl *http.Client, projects []string) ([]s
 	return ret, nil
 }
 
-func gcloudProject() []string {
-	buf := new(bytes.Buffer)
-	cmd := exec.Command("gcloud", "--format", "json", "config", "list", "core/project")
-	cmd.Stdout = buf
-
-	if err := cmd.Run(); err != nil {
-		if strings.Contains(err.Error(), "executable file not found") {
-			// gcloud not installed; ignore the error
-			return nil
-		}
-		logging.Errorf("Error detecting gcloud project: %v", err)
-		return nil
-	}
-
-	var data struct {
-		Core struct {
-			Project string
-		}
+func gcloudProject() ([]string, error) {
+	cfg, err := util.GcloudConfig()
+	if err != nil {
+		return nil, err
 	}
-
-	if err := json.Unmarshal(buf.Bytes(), &data); err != nil {
-		logging.Errorf("Failed to unmarshal bytes from gcloud: %v", err)
-		logging.Errorf("   gcloud returned:\n%s", buf)
-		return nil
+	if cfg.Configuration.Properties.Core.Project == "" {
+		return nil, fmt.Errorf("gcloud has no active project, you can set it by running `gcloud config set project <project>`")
 	}
-
-	logging.Infof("Using gcloud's active project: %v", data.Core.Project)
-	return []string{data.Core.Project}
+	return []string{cfg.Configuration.Properties.Core.Project}, nil
 }
 
 // Main executes the main function of the proxy, allowing it to be called from tests.
@@ -392,6 +385,15 @@ func main() {
 		log.SetOutput(ioutil.Discard)
 	}
 
+	// Split the input ipAddressTypes to the slice of string
+	ipAddrTypeOptsInput := strings.Split(*ipAddressTypes, ",")
+
+	if *fdRlimit != 0 {
+		if err := limits.SetupFDLimits(*fdRlimit); err != nil {
+			logging.Infof("failed to setup file descriptor limits: %v", err)
+		}
+	}
+
 	// TODO: needs a better place for consolidation
 	// if instances is blank and env var INSTANCES is supplied use it
 	if envInstances := os.Getenv("INSTANCES"); *instances == "" && envInstances != "" {
@@ -402,7 +404,15 @@ func main() {
 	projList := stringList(*projects)
 	// TODO: it'd be really great to consolidate flag verification in one place.
 	if len(instList) == 0 && *instanceSrc == "" && len(projList) == 0 && !*useFuse {
-		projList = gcloudProject()
+		var err error
+		projList, err = gcloudProject()
+		if err == nil {
+			logging.Infof("Using gcloud's active project: %v", projList)
+		} else if gErr, ok := err.(*util.GcloudError); ok && gErr.Status == util.GcloudNotFound {
+			log.Fatalf("gcloud is not in the path and -instances and -projects are empty")
+		} else {
+			log.Fatalf("unable to retrieve the active gcloud project and -instances and -projects are empty: %v", err)
+		}
 	}
 
 	onGCE := onGCE()
@@ -421,7 +431,6 @@ func main() {
 		log.Fatal(err)
 	}
 	instList = append(instList, ins...)
-
 	cfgs, err := CreateInstanceConfigs(*dir, *useFuse, instList, *instanceSrc, client)
 	if err != nil {
 		log.Fatal(err)
@@ -477,9 +486,10 @@ func main() {
 		Port:           port,
 		MaxConnections: *maxConnections,
 		Certs: certs.NewCertSourceOpts(client, certs.RemoteOpts{
-			APIBasePath:  host,
-			IgnoreRegion: !*checkRegion,
-			UserAgent:    userAgentFromVersionString(),
+			APIBasePath:    *host,
+			IgnoreRegion:   !*checkRegion,
+			UserAgent:      userAgentFromVersionString(),
+			IPAddrTypeOpts: ipAddrTypeOptsInput,
 		}),
 		Conns:              connset,
 		RefreshCfgThrottle: refreshCfgThrottle,

cmd/cloud_sql_proxy/proxy.go

@@ -155,6 +155,13 @@ func listenInstance(dst chan<- proxy.Conn, cfg instanceConfig) (net.Listener, er
 				return
 			}
 			logging.Verbosef("New connection for %q", cfg.Instance)
+
+			switch clientConn := c.(type) {
+			case *net.TCPConn:
+				clientConn.SetKeepAlive(true)
+				clientConn.SetKeepAlivePeriod(1 * time.Minute)
+
+			}
 			dst <- proxy.Conn{cfg.Instance, c}
 		}
 	}()
@@ -241,7 +248,7 @@ func parseInstanceConfig(dir, instance string, cl *http.Client) (instanceConfig,
 		if err != nil {
 			return instanceConfig{}, err
 		}
-
+		sql.BasePath = *host
 		ret.Instance = instance
 		// Default to unix socket.
 		ret.Network = "unix"
@@ -338,10 +345,18 @@ func CreateInstanceConfigs(dir string, useFuse bool, instances []string, instanc
 	}
 	// FUSE disabled.
 	if len(instances) == 0 && instancesSrc == "" {
+		// Failure to specifying instance can be caused by following reasons.
+		// 1. not enough information is provided by flags
+		// 2. failed to invoke gcloud
+		var flags string
 		if fuse.Supported() {
-			return nil, errors.New("must specify -projects, -fuse, or -instances")
+			flags = "-projects, -fuse, -instances or -instances_metadata"
+		} else {
+			flags = "-projects, -instances or -instances_metadata"
 		}
-		return nil, errors.New("must specify -projects or -instances")
+
+		errStr := fmt.Sprintf("no instance selected because none of %s is specified", flags)
+		return nil, errors.New(errStr)
 	}
 	return cfgs, nil
 }

proxy/certs/certs.go

@@ -26,6 +26,7 @@ import (
 	"math"
 	mrand "math/rand"
 	"net/http"
+	"strings"
 	"time"
 
 	"github.com/GoogleCloudPlatform/cloudsql-proxy/logging"
@@ -64,6 +65,9 @@ type RemoteOpts struct {
 	// A string for the RemoteCertSource to identify itself when contacting the
 	// sqladmin API.
 	UserAgent string
+
+	// IP address type options
+	IPAddrTypeOpts []string
 }
 
 // NewCertSourceOpts returns a CertSource configured with the provided Opts.
@@ -88,7 +92,20 @@ func NewCertSourceOpts(c *http.Client, opts RemoteOpts) *RemoteCertSource {
 		ua = defaultUserAgent
 	}
 	serv.UserAgent = ua
-	return &RemoteCertSource{pkey, serv, !opts.IgnoreRegion}
+
+	// Set default value to be PRIMARY if input opts.IPAddrTypeOpts is empty
+	if len(opts.IPAddrTypeOpts) < 1 {
+		opts.IPAddrTypeOpts = append(opts.IPAddrTypeOpts, "PRIMARY")
+	} else {
+		// Add "PUBLIC" as an alias for "PRIMARY"
+		for index, ipAddressType := range opts.IPAddrTypeOpts {
+			if strings.ToUpper(ipAddressType) == "PUBLIC" {
+				opts.IPAddrTypeOpts[index] = "PRIMARY"
+			}
+		}
+	}
+
+	return &RemoteCertSource{pkey, serv, !opts.IgnoreRegion, opts.IPAddrTypeOpts}
 }
 
 // RemoteCertSource implements a CertSource, using Cloud SQL APIs to
@@ -104,6 +121,8 @@ type RemoteCertSource struct {
 	// treated as an error. This is to provide the same functionality that will
 	// occur when API calls require the region.
 	checkRegion bool
+	// a list of ip address types that users select
+	IPAddrTypes []string
 }
 
 // Constants for backoffAPIRetry. These cause the retry logic to scale the
@@ -185,6 +204,27 @@ func parseCert(pemCert string) (*x509.Certificate, error) {
 	return x509.ParseCertificate(bl.Bytes)
 }
 
+// Find the first matching IP address by user input IP address types
+func (s *RemoteCertSource) findIPAddr(data *sqladmin.DatabaseInstance, instance string) (ipAddrInUse string, err error) {
+	for _, eachIPAddrTypeByUser := range s.IPAddrTypes {
+		for _, eachIPAddrTypeOfInstance := range data.IpAddresses {
+			if strings.ToUpper(eachIPAddrTypeOfInstance.Type) == strings.ToUpper(eachIPAddrTypeByUser) {
+				ipAddrInUse = eachIPAddrTypeOfInstance.IpAddress
+				return ipAddrInUse, nil
+			}
+		}
+	}
+
+	ipAddrTypesOfInstance := ""
+	for _, eachIPAddrTypeOfInstance := range data.IpAddresses {
+		ipAddrTypesOfInstance += fmt.Sprintf("(TYPE=%v, IP_ADDR=%v)", eachIPAddrTypeOfInstance.Type, eachIPAddrTypeOfInstance.IpAddress)
+	}
+
+	ipAddrTypeOfUser := fmt.Sprintf("%v", s.IPAddrTypes)
+
+	return "", fmt.Errorf("User input IP address type %v does not match the instance %v, the instance's IP addresses are %v ", ipAddrTypeOfUser, instance, ipAddrTypesOfInstance)
+}
+
 // Remote returns the specified instance's CA certificate, address, and name.
 func (s *RemoteCertSource) Remote(instance string) (cert *x509.Certificate, addr, name string, err error) {
 	p, region, n := util.SplitName(instance)
@@ -215,9 +255,19 @@ func (s *RemoteCertSource) Remote(instance string) (cert *x509.Certificate, addr
 		logging.Errorf("%v", err)
 		logging.Errorf("WARNING: specifying the correct region in an instance string will become required in a future version!")
 	}
-	if len(data.IpAddresses) == 0 || data.IpAddresses[0].IpAddress == "" {
+
+	if len(data.IpAddresses) == 0 {
 		return nil, "", "", fmt.Errorf("no IP address found for %v", instance)
 	}
+
+	// Find the first matching IP address by user input IP address types
+	ipAddrInUse := ""
+	ipAddrInUse, err = s.findIPAddr(data, instance)
+	if err != nil {
+		return nil, "", "", err
+	}
+
 	c, err := parseCert(data.ServerCaCert.Cert)
-	return c, data.IpAddresses[0].IpAddress, p + ":" + n, err
+
+	return c, ipAddrInUse, p + ":" + n, err
 }

proxy/dialers/mysql/hook.go

@@ -38,12 +38,10 @@ func init() {
 // The returned *sql.DB may be valid even if there's also an error returned
 // (e.g. if there was a transient connection error).
 func Dial(instance, user string) (*sql.DB, error) {
-	return DialCfg(&mysql.Config{
-		User: user,
-		Addr: instance,
-		// Set in DialCfg:
-		// Net: "cloudsql",
-	})
+	cfg := mysql.NewConfig()
+	cfg.User = user
+	cfg.Addr = instance
+	return DialCfg(cfg)
 }
 
 // DialPassword is similar to Dial, but allows you to specify a password.
@@ -53,26 +51,24 @@ func Dial(instance, user string) (*sql.DB, error) {
 // information, see:
 //    https://cloud.google.com/sql/docs/sql-proxy#user
 func DialPassword(instance, user, password string) (*sql.DB, error) {
-	return DialCfg(&mysql.Config{
-		User:   user,
-		Passwd: password,
-		Addr:   instance,
-		// Set in DialCfg:
-		// Net: "cloudsql",
-	})
+	cfg := mysql.NewConfig()
+	cfg.User = user
+	cfg.Passwd = password
+	cfg.Addr = instance
+	return DialCfg(cfg)
 }
 
 // Cfg returns the effective *mysql.Config to represent connectivity to the
 // provided instance via the given user and password. The config can be
 // modified and passed to DialCfg to connect. If you don't modify the returned
 // config before dialing, consider using Dial or DialPassword.
 func Cfg(instance, user, password string) *mysql.Config {
-	return &mysql.Config{
-		Addr:   instance,
-		User:   user,
-		Passwd: password,
-		Net:    "cloudsql",
-	}
+	cfg := mysql.NewConfig()
+	cfg.User = user
+	cfg.Passwd = password
+	cfg.Addr = instance
+	cfg.Net = "cloudsql"
+	return cfg
 }
 
 // DialCfg opens up a SQL connection to a Cloud SQL Instance specified by the