Commit a6c43c6a authored by hzxuzhonghu's avatar hzxuzhonghu

pass listener in integration test to prevent port in use flake

parent 84408378
......@@ -35,7 +35,8 @@ import (
type SecureServingOptions struct {
BindAddress net.IP
BindPort int
// BindPort is ignored when Listener is set, will serve https even with 0.
BindPort int
// BindNetwork is the type of network to bind to - defaults to "tcp", accepts "tcp",
// "tcp4", and "tcp6".
BindNetwork string
......@@ -160,7 +161,7 @@ func (s *SecureServingOptions) ApplyTo(c *server.Config) error {
if s == nil {
return nil
}
if s.BindPort <= 0 {
if s.BindPort <= 0 && s.Listener == nil {
return nil
}
......@@ -171,6 +172,12 @@ func (s *SecureServingOptions) ApplyTo(c *server.Config) error {
if err != nil {
return fmt.Errorf("failed to create listener: %v", err)
}
} else {
if _, ok := s.Listener.Addr().(*net.TCPAddr); !ok {
return fmt.Errorf("failed to parse ip and port from listener")
}
s.BindPort = s.Listener.Addr().(*net.TCPAddr).Port
s.BindAddress = s.Listener.Addr().(*net.TCPAddr).IP
}
if err := s.applyServingInfoTo(c); err != nil {
......
......@@ -44,7 +44,8 @@ func main() {
}
stopCh := genericapiserver.SetupSignalHandler()
cmd := server.NewCommandStartAggregator(os.Stdout, os.Stderr, stopCh)
options := server.NewDefaultOptions(os.Stdout, os.Stderr)
cmd := server.NewCommandStartAggregator(options, stopCh)
cmd.Flags().AddGoFlagSet(flag.CommandLine)
if err := cmd.Execute(); err != nil {
glog.Fatal(err)
......
......@@ -49,9 +49,9 @@ type AggregatorOptions struct {
}
// NewCommandStartAggregator provides a CLI handler for 'start master' command
func NewCommandStartAggregator(out, err io.Writer, stopCh <-chan struct{}) *cobra.Command {
o := NewDefaultOptions(out, err)
// with a default AggregatorOptions.
func NewCommandStartAggregator(defaults *AggregatorOptions, stopCh <-chan struct{}) *cobra.Command {
o := *defaults
cmd := &cobra.Command{
Short: "Launch a API aggregator and proxy server",
Long: "Launch a API aggregator and proxy server",
......
......@@ -37,7 +37,8 @@ func main() {
}
stopCh := genericapiserver.SetupSignalHandler()
cmd := server.NewCommandStartWardleServer(os.Stdout, os.Stderr, stopCh)
options := server.NewWardleServerOptions(os.Stdout, os.Stderr)
cmd := server.NewCommandStartWardleServer(options, stopCh)
cmd.Flags().AddGoFlagSet(flag.CommandLine)
if err := cmd.Execute(); err != nil {
glog.Fatal(err)
......
......@@ -57,9 +57,9 @@ func NewWardleServerOptions(out, errOut io.Writer) *WardleServerOptions {
}
// NewCommandStartWardleServer provides a CLI handler for 'start master' command
func NewCommandStartWardleServer(out, errOut io.Writer, stopCh <-chan struct{}) *cobra.Command {
o := NewWardleServerOptions(out, errOut)
// with a default WardleServerOptions.
func NewCommandStartWardleServer(defaults *WardleServerOptions, stopCh <-chan struct{}) *cobra.Command {
o := *defaults
cmd := &cobra.Command{
Short: "Launch a wardle API server",
Long: "Launch a wardle API server",
......
......@@ -41,6 +41,7 @@ import (
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apimachinery/pkg/util/wait"
genericapiserver "k8s.io/apiserver/pkg/server"
genericapiserveroptions "k8s.io/apiserver/pkg/server/options"
"k8s.io/apiserver/pkg/storage/storagebackend"
clientset "k8s.io/client-go/kubernetes"
restclient "k8s.io/client-go/rest"
......@@ -704,47 +705,41 @@ func startRealMasterOrDie(t *testing.T, certDir string) (*allClient, clientv3.KV
}
}()
for {
kubeAPIServerOptions := options.NewServerRunOptions()
kubeAPIServerOptions.SecureServing.BindAddress = net.ParseIP("127.0.0.1")
kubeAPIServerOptions.SecureServing.ServerCert.CertDirectory = certDir
kubeAPIServerOptions.Etcd.StorageConfig.ServerList = []string{framework.GetEtcdURL()}
kubeAPIServerOptions.Etcd.DefaultStorageMediaType = runtime.ContentTypeJSON // TODO use protobuf?
kubeAPIServerOptions.ServiceClusterIPRange = *defaultServiceClusterIPRange
kubeAPIServerOptions.Authorization.Mode = "RBAC"
// always get a fresh port in case something claimed the old one
kubePort, err := framework.FindFreeLocalPort()
if err != nil {
t.Fatal(err)
}
listener, _, err := genericapiserveroptions.CreateListener("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
kubeAPIServerOptions.SecureServing.BindPort = kubePort
kubeAPIServerOptions := options.NewServerRunOptions()
tunneler, proxyTransport, err := app.CreateNodeDialer(kubeAPIServerOptions)
if err != nil {
t.Fatal(err)
}
kubeAPIServerConfig, sharedInformers, versionedInformers, _, _, err := app.CreateKubeAPIServerConfig(kubeAPIServerOptions, tunneler, proxyTransport)
if err != nil {
t.Fatal(err)
}
kubeAPIServerOptions.SecureServing.Listener = listener
kubeAPIServerOptions.SecureServing.ServerCert.CertDirectory = certDir
kubeAPIServerOptions.Etcd.StorageConfig.ServerList = []string{framework.GetEtcdURL()}
kubeAPIServerOptions.Etcd.DefaultStorageMediaType = runtime.ContentTypeJSON // TODO use protobuf?
kubeAPIServerOptions.ServiceClusterIPRange = *defaultServiceClusterIPRange
kubeAPIServerOptions.Authorization.Mode = "RBAC"
kubeAPIServerConfig.ExtraConfig.APIResourceConfigSource = &allResourceSource{} // force enable all resources
tunneler, proxyTransport, err := app.CreateNodeDialer(kubeAPIServerOptions)
if err != nil {
t.Fatal(err)
}
kubeAPIServerConfig, sharedInformers, versionedInformers, _, _, err := app.CreateKubeAPIServerConfig(kubeAPIServerOptions, tunneler, proxyTransport)
if err != nil {
t.Fatal(err)
}
kubeAPIServer, err := app.CreateKubeAPIServer(kubeAPIServerConfig, genericapiserver.EmptyDelegate, sharedInformers, versionedInformers)
if err != nil {
t.Fatal(err)
}
kubeAPIServerConfig.ExtraConfig.APIResourceConfigSource = &allResourceSource{} // force enable all resources
kubeClientConfigValue.Store(kubeAPIServerConfig.GenericConfig.LoopbackClientConfig)
storageConfigValue.Store(kubeAPIServerOptions.Etcd.StorageConfig)
kubeAPIServer, err := app.CreateKubeAPIServer(kubeAPIServerConfig, genericapiserver.EmptyDelegate, sharedInformers, versionedInformers)
if err != nil {
t.Fatal(err)
}
if err := kubeAPIServer.GenericAPIServer.PrepareRun().Run(wait.NeverStop); err != nil {
t.Log(err)
}
kubeClientConfigValue.Store(kubeAPIServerConfig.GenericConfig.LoopbackClientConfig)
storageConfigValue.Store(kubeAPIServerOptions.Etcd.StorageConfig)
time.Sleep(time.Second)
if err := kubeAPIServer.GenericAPIServer.PrepareRun().Run(wait.NeverStop); err != nil {
t.Fatal(err)
}
}()
......
......@@ -21,7 +21,6 @@ import (
"net/http"
"net/http/httptest"
"path"
"strconv"
"time"
"github.com/go-openapi/spec"
......@@ -328,28 +327,6 @@ func RunAMasterUsingServer(masterConfig *master.Config, s *httptest.Server, mast
return startMasterOrDie(masterConfig, s, masterReceiver)
}
// FindFreeLocalPort returns the number of an available port number on
// the loopback interface. Useful for determining the port to launch
// a server on. Error handling required - there is a non-zero chance
// that the returned port number will be bound by another process
// after this function returns.
func FindFreeLocalPort() (int, error) {
l, err := net.Listen("tcp", ":0")
if err != nil {
return 0, err
}
defer l.Close()
_, portStr, err := net.SplitHostPort(l.Addr().String())
if err != nil {
return 0, err
}
port, err := strconv.Atoi(portStr)
if err != nil {
return 0, err
}
return port, nil
}
// SharedEtcd creates a storage config for a shared etcd instance, with a unique prefix.
func SharedEtcd() *storagebackend.Config {
cfg := storagebackend.NewDefaultConfig(path.Join(uuid.New(), "registry"), nil)
......
......@@ -29,6 +29,7 @@ import (
"k8s.io/apimachinery/pkg/util/wait"
genericapiserver "k8s.io/apiserver/pkg/server"
genericapiserveroptions "k8s.io/apiserver/pkg/server/options"
client "k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/kubernetes/cmd/kube-apiserver/app"
......@@ -44,17 +45,17 @@ func runBasicSecureAPIServer(t *testing.T, ciphers []string) (uint32, error) {
var kubePort uint32
go func() {
// always get a fresh port in case something claimed the old one
freePort, err := framework.FindFreeLocalPort()
listener, port, err := genericapiserveroptions.CreateListener("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
atomic.StoreUint32(&kubePort, uint32(freePort))
atomic.StoreUint32(&kubePort, uint32(port))
kubeAPIServerOptions := options.NewServerRunOptions()
kubeAPIServerOptions.SecureServing.BindAddress = net.ParseIP("127.0.0.1")
kubeAPIServerOptions.SecureServing.BindPort = freePort
kubeAPIServerOptions.SecureServing.BindPort = port
kubeAPIServerOptions.SecureServing.Listener = listener
kubeAPIServerOptions.SecureServing.ServerCert.CertDirectory = certDir
kubeAPIServerOptions.SecureServing.CipherSuites = ciphers
kubeAPIServerOptions.InsecureServing.BindPort = 0
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment