Merge remote-tracking branch 'origin/master' into release-1.13. Deleting…

cluster/addons/fluentd-gcp/fluentd-gcp-configmap-old.yaml

@@ -61,7 +61,8 @@ data:
       # reform.var.log.containers.<POD_NAME>_<NAMESPACE_NAME>_<CONTAINER_NAME>-<CONTAINER_ID>.log
       tag reform.*
       read_from_head true
-      format multi_format
+      <parse>
+        @type multi_format
         <pattern>
           format json
           time_key time
@@ -71,6 +72,7 @@ data:
           format /^(?<time>.+) (?<stream>stdout|stderr) [^ ]* (?<log>.*)$/
           time_format %Y-%m-%dT%H:%M:%S.%N%:z
         </pattern>
+      </parse>
     </source>
 
     <filter reform.**>

cluster/addons/fluentd-gcp/fluentd-gcp-configmap.yaml

@@ -64,7 +64,8 @@ data:
       # reform.var.log.containers.<POD_NAME>_<NAMESPACE_NAME>_<CONTAINER_NAME>-<CONTAINER_ID>.log
       tag reform.*
       read_from_head true
-      format multi_format
+      <parse>
+        @type multi_format
         <pattern>
           format json
           time_key time
@@ -74,6 +75,7 @@ data:
           format /^(?<time>.+) (?<stream>stdout|stderr) [^ ]* (?<log>.*)$/
           time_format %Y-%m-%dT%H:%M:%S.%N%:z
         </pattern>
+      </parse>
     </source>
 
     <filter reform.**>

cluster/gce/config-default.sh

@@ -409,8 +409,8 @@ fi
 
 # Fluentd requirements
 # YAML exists to trigger a configuration refresh when changes are made.
-FLUENTD_GCP_YAML_VERSION="v3.1.0"
-FLUENTD_GCP_VERSION="${FLUENTD_GCP_VERSION:-0.5-1.5.36-1-k8s}"
+FLUENTD_GCP_YAML_VERSION="v3.2.0"
+FLUENTD_GCP_VERSION="${FLUENTD_GCP_VERSION:-0.6-1.6.0-1}"
 FLUENTD_GCP_MEMORY_LIMIT="${FLUENTD_GCP_MEMORY_LIMIT:-}"
 FLUENTD_GCP_CPU_REQUEST="${FLUENTD_GCP_CPU_REQUEST:-}"
 FLUENTD_GCP_MEMORY_REQUEST="${FLUENTD_GCP_MEMORY_REQUEST:-}"

cluster/gce/config-test.sh

@@ -425,8 +425,8 @@ fi
 
 # Fluentd requirements
 # YAML exists to trigger a configuration refresh when changes are made.
-FLUENTD_GCP_YAML_VERSION="v3.1.0"
-FLUENTD_GCP_VERSION="${FLUENTD_GCP_VERSION:-0.5-1.5.36-1-k8s}"
+FLUENTD_GCP_YAML_VERSION="v3.2.0"
+FLUENTD_GCP_VERSION="${FLUENTD_GCP_VERSION:-0.6-1.6.0-1}"
 FLUENTD_GCP_MEMORY_LIMIT="${FLUENTD_GCP_MEMORY_LIMIT:-}"
 FLUENTD_GCP_CPU_REQUEST="${FLUENTD_GCP_CPU_REQUEST:-}"
 FLUENTD_GCP_MEMORY_REQUEST="${FLUENTD_GCP_MEMORY_REQUEST:-}"

cluster/gce/gci/configure-helper.sh

@@ -2321,10 +2321,10 @@ function setup-fluentd {
     fluentd_gcp_configmap_name="fluentd-gcp-config-old"
   fi
   sed -i -e "s@{{ fluentd_gcp_configmap_name }}@${fluentd_gcp_configmap_name}@g" "${fluentd_gcp_yaml}"
-  fluentd_gcp_yaml_version="${FLUENTD_GCP_YAML_VERSION:-v3.1.0}"
+  fluentd_gcp_yaml_version="${FLUENTD_GCP_YAML_VERSION:-v3.2.0}"
   sed -i -e "s@{{ fluentd_gcp_yaml_version }}@${fluentd_gcp_yaml_version}@g" "${fluentd_gcp_yaml}"
   sed -i -e "s@{{ fluentd_gcp_yaml_version }}@${fluentd_gcp_yaml_version}@g" "${fluentd_gcp_scaler_yaml}"
-  fluentd_gcp_version="${FLUENTD_GCP_VERSION:-0.5-1.5.36-1-k8s}"
+  fluentd_gcp_version="${FLUENTD_GCP_VERSION:-0.6-1.6.0-1}"
   sed -i -e "s@{{ fluentd_gcp_version }}@${fluentd_gcp_version}@g" "${fluentd_gcp_yaml}"
   update-daemon-set-prometheus-to-sd-parameters ${fluentd_gcp_yaml}
   start-fluentd-resource-update ${fluentd_gcp_yaml}

staging/src/k8s.io/apimachinery/pkg/util/proxy/upgradeaware.go

@@ -17,6 +17,7 @@ limitations under the License.
 package proxy
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"fmt"
@@ -271,6 +272,18 @@ func (h *UpgradeAwareHandler) tryUpgrade(w http.ResponseWriter, req *http.Reques
 	}
 	defer backendConn.Close()
 
+	// determine the http response code from the backend by reading from rawResponse+backendConn
+	backendHTTPResponse, headerBytes, err := getResponse(io.MultiReader(bytes.NewReader(rawResponse), backendConn))
+	if err != nil {
+		klog.V(6).Infof("Proxy connection error: %v", err)
+		h.Responder.Error(w, req, err)
+		return true
+	}
+	if len(headerBytes) > len(rawResponse) {
+		// we read beyond the bytes stored in rawResponse, update rawResponse to the full set of bytes read from the backend
+		rawResponse = headerBytes
+	}
+
 	// Once the connection is hijacked, the ErrorResponder will no longer work, so
 	// hijacking should be the last step in the upgrade.
 	requestHijacker, ok := w.(http.Hijacker)
@@ -287,6 +300,22 @@ func (h *UpgradeAwareHandler) tryUpgrade(w http.ResponseWriter, req *http.Reques
 	}
 	defer requestHijackedConn.Close()
 
+	if backendHTTPResponse.StatusCode != http.StatusSwitchingProtocols {
+		// If the backend did not upgrade the request, echo the response from the backend to the client and return, closing the connection.
+		klog.V(6).Infof("Proxy upgrade error, status code %d", backendHTTPResponse.StatusCode)
+		// set read/write deadlines
+		deadline := time.Now().Add(10 * time.Second)
+		backendConn.SetReadDeadline(deadline)
+		requestHijackedConn.SetWriteDeadline(deadline)
+		// write the response to the client
+		err := backendHTTPResponse.Write(requestHijackedConn)
+		if err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
+			klog.Errorf("Error proxying data from backend to client: %v", err)
+		}
+		// Indicate we handled the request
+		return true
+	}
+
 	// Forward raw response bytes back to client.
 	if len(rawResponse) > 0 {
 		klog.V(6).Infof("Writing %d bytes to hijacked connection", len(rawResponse))
@@ -356,6 +385,19 @@ func (h *UpgradeAwareHandler) DialForUpgrade(req *http.Request) (net.Conn, error
 	return dial(updatedReq, h.UpgradeTransport)
 }
 
+// getResponseCode reads a http response from the given reader, returns the response,
+// the bytes read from the reader, and any error encountered
+func getResponse(r io.Reader) (*http.Response, []byte, error) {
+	rawResponse := bytes.NewBuffer(make([]byte, 0, 256))
+	// Save the bytes read while reading the response headers into the rawResponse buffer
+	resp, err := http.ReadResponse(bufio.NewReader(io.TeeReader(r, rawResponse)), nil)
+	if err != nil {
+		return nil, nil, err
+	}
+	// return the http response and the raw bytes consumed from the reader in the process
+	return resp, rawResponse.Bytes(), nil
+}
+
 // dial dials the backend at req.URL and writes req to it.
 func dial(req *http.Request, transport http.RoundTripper) (net.Conn, error) {
 	conn, err := DialURL(req.Context(), req.URL, transport)

test/e2e/framework/util.go

@@ -1496,10 +1496,12 @@ func eventOccurred(c clientset.Interface, podName, namespace, eventSelector, msg
 		if err != nil {
 			return false, fmt.Errorf("got error while getting pod events: %s", err)
 		}
-		if len(events.Items) == 0 {
-			return false, nil // no events have occurred yet
+		for _, event := range events.Items {
+			if strings.Contains(event.Message, msg) {
+				return true, nil
+			}
 		}
-		return strings.Contains(events.Items[0].Message, msg), nil
+		return false, nil
 	}
 }
 

test/e2e/storage/detach_mounted.go

@@ -84,10 +84,17 @@ var _ = utils.SIGDescribe("Detaching volumes", func() {
 
 		uniqueVolumeName := getUniqueVolumeName(pod, driverInstallAs)
 
+		By("waiting for volumes to be attached to node")
+		err = waitForVolumesAttached(cs, node.Name, uniqueVolumeName)
+		Expect(err).NotTo(HaveOccurred(), "while waiting for volume to attach to %s node", node.Name)
+
 		By("waiting for volume-in-use on the node after pod creation")
 		err = waitForVolumesInUse(cs, node.Name, uniqueVolumeName)
 		Expect(err).NotTo(HaveOccurred(), "while waiting for volume in use")
 
+		By("waiting for kubelet to start mounting the volume")
+		time.Sleep(20 * time.Second)
+
 		By("Deleting the flexvolume pod")
 		err = framework.DeletePodWithWait(f, cs, pod)
 		Expect(err).NotTo(HaveOccurred(), "in deleting the pod")
@@ -134,6 +141,22 @@ func waitForVolumesNotInUse(client clientset.Interface, nodeName, volumeName str
 	})
 }
 
+func waitForVolumesAttached(client clientset.Interface, nodeName, volumeName string) error {
+	return wait.PollImmediate(2*time.Second, 2*time.Minute, func() (bool, error) {
+		node, err := client.CoreV1().Nodes().Get(nodeName, metav1.GetOptions{})
+		if err != nil {
+			return false, fmt.Errorf("error fetching node %s with %v", nodeName, err)
+		}
+		volumeAttached := node.Status.VolumesAttached
+		for _, volume := range volumeAttached {
+			if string(volume.Name) == volumeName {
+				return true, nil
+			}
+		}
+		return false, nil
+	})
+}
+
 func waitForVolumesInUse(client clientset.Interface, nodeName, volumeName string) error {
 	return wait.PollImmediate(10*time.Second, 60*time.Second, func() (bool, error) {
 		node, err := client.CoreV1().Nodes().Get(nodeName, metav1.GetOptions{})

test/e2e/storage/testsuites/subpath.go

@@ -587,8 +587,6 @@ func testPodFailSubpathError(f *framework.Framework, pod *v1.Pod, errorMsg strin
 	defer func() {
 		framework.DeletePodWithWait(f, f.ClientSet, pod)
 	}()
-	err = framework.WaitForPodRunningInNamespace(f.ClientSet, pod)
-	Expect(err).To(HaveOccurred(), "while waiting for pod to be running")
 
 	By("Checking for subpath error event")
 	selector := fields.Set{

test/e2e/storage/testsuites/volume_io.go

@@ -24,7 +24,7 @@ package testsuites
 import (
 	"fmt"
 	"math"
-	"path"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"time"
@@ -49,6 +49,8 @@ var md5hashes = map[int64]string{
 	testpatterns.FileSizeLarge:  "8d763edc71bd16217664793b5a15e403",
 }
 
+const mountPath = "/opt"
+
 type volumeIOTestSuite struct {
 	tsInfo TestSuiteInfo
 }
@@ -176,10 +178,9 @@ func createFileSizes(maxFileSize int64) []int64 {
 }
 
 // Return the plugin's client pod spec. Use an InitContainer to setup the file i/o test env.
-func makePodSpec(config framework.VolumeTestConfig, dir, initCmd string, volsrc v1.VolumeSource, podSecContext *v1.PodSecurityContext) *v1.Pod {
-	volName := fmt.Sprintf("%s-%s", config.Prefix, "io-volume")
-
+func makePodSpec(config framework.VolumeTestConfig, initCmd string, volsrc v1.VolumeSource, podSecContext *v1.PodSecurityContext) *v1.Pod {
 	var gracePeriod int64 = 1
+	volName := fmt.Sprintf("io-volume-%s", config.Namespace)
 	return &v1.Pod{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "Pod",
@@ -204,7 +205,7 @@ func makePodSpec(config framework.VolumeTestConfig, dir, initCmd string, volsrc
 					VolumeMounts: []v1.VolumeMount{
 						{
 							Name:      volName,
-							MountPath: dir,
+							MountPath: mountPath,
 						},
 					},
 				},
@@ -221,7 +222,7 @@ func makePodSpec(config framework.VolumeTestConfig, dir, initCmd string, volsrc
 					VolumeMounts: []v1.VolumeMount{
 						{
 							Name:      volName,
-							MountPath: dir,
+							MountPath: mountPath,
 						},
 					},
 				},
@@ -302,8 +303,7 @@ func deleteFile(pod *v1.Pod, fpath string) {
 // Note: `fsizes` values are enforced to each be at least `MinFileSize` and a multiple of `MinFileSize`
 //   bytes.
 func testVolumeIO(f *framework.Framework, cs clientset.Interface, config framework.VolumeTestConfig, volsrc v1.VolumeSource, podSecContext *v1.PodSecurityContext, file string, fsizes []int64) (err error) {
-	dir := path.Join("/opt", config.Prefix, config.Namespace)
-	ddInput := path.Join(dir, "dd_if")
+	ddInput := filepath.Join(mountPath, fmt.Sprintf("%s-%s-dd_if", config.Prefix, config.Namespace))
 	writeBlk := strings.Repeat("abcdefghijklmnopqrstuvwxyz123456", 32) // 1KiB value
 	loopCnt := testpatterns.MinFileSize / int64(len(writeBlk))
 	// initContainer cmd to create and fill dd's input file. The initContainer is used to create
@@ -311,7 +311,7 @@ func testVolumeIO(f *framework.Framework, cs clientset.Interface, config framewo
 	// used to create a 1MiB file in the target directory.
 	initCmd := fmt.Sprintf("i=0; while [ $i -lt %d ]; do echo -n %s >>%s; let i+=1; done", loopCnt, writeBlk, ddInput)
 
-	clientPod := makePodSpec(config, dir, initCmd, volsrc, podSecContext)
+	clientPod := makePodSpec(config, initCmd, volsrc, podSecContext)
 
 	By(fmt.Sprintf("starting %s", clientPod.Name))
 	podsNamespacer := cs.CoreV1().Pods(config.Namespace)
@@ -320,7 +320,7 @@ func testVolumeIO(f *framework.Framework, cs clientset.Interface, config framewo
 		return fmt.Errorf("failed to create client pod %q: %v", clientPod.Name, err)
 	}
 	defer func() {
-		// note the test dir will be removed when the kubelet unmounts it
+		deleteFile(clientPod, ddInput)
 		By(fmt.Sprintf("deleting client pod %q...", clientPod.Name))
 		e := framework.DeletePodWithWait(f, cs, clientPod)
 		if e != nil {
@@ -345,14 +345,16 @@ func testVolumeIO(f *framework.Framework, cs clientset.Interface, config framewo
 		if math.Mod(float64(fsize), float64(testpatterns.MinFileSize)) != 0 {
 			fsize = fsize/testpatterns.MinFileSize + testpatterns.MinFileSize
 		}
-		fpath := path.Join(dir, fmt.Sprintf("%s-%d", file, fsize))
+		fpath := filepath.Join(mountPath, fmt.Sprintf("%s-%d", file, fsize))
+		defer func() {
+			deleteFile(clientPod, fpath)
+		}()
 		if err = writeToFile(clientPod, fpath, ddInput, fsize); err != nil {
 			return err
 		}
 		if err = verifyFile(clientPod, fpath, fsize, ddInput); err != nil {
 			return err
 		}
-		deleteFile(clientPod, fpath)
 	}
 
 	return