grpc_test.go

Documentation: k8s.io/kubernetes/pkg/probe/grpc

     1  /*
     2  Copyright 2021 The Kubernetes Authors.
     3  
     4  Licensed under the Apache License, Version 2.0 (the "License");
     5  you may not use this file except in compliance with the License.
     6  You may obtain a copy of the License at
     7  
     8      http://www.apache.org/licenses/LICENSE-2.0
     9  
    10  Unless required by applicable law or agreed to in writing, software
    11  distributed under the License is distributed on an "AS IS" BASIS,
    12  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13  See the License for the specific language governing permissions and
    14  limitations under the License.
    15  */
    16  
    17  package grpc
    18  
    19  import (
    20  	"context"
    21  	"fmt"
    22  	"net"
    23  	"net/http"
    24  	"net/http/httptest"
    25  	"strconv"
    26  	"strings"
    27  	"testing"
    28  	"time"
    29  
    30  	"github.com/stretchr/testify/assert"
    31  	"google.golang.org/grpc"
    32  	grpchealth "google.golang.org/grpc/health/grpc_health_v1"
    33  
    34  	"k8s.io/kubernetes/pkg/probe"
    35  )
    36  
    37  func TestNew(t *testing.T) {
    38  	t.Run("Should: implement Probe interface", func(t *testing.T) {
    39  		s := New()
    40  		assert.Implements(t, (*Prober)(nil), s)
    41  	})
    42  }
    43  
    44  type successServerMock struct {
    45  }
    46  
    47  func (s successServerMock) Check(context.Context, *grpchealth.HealthCheckRequest) (*grpchealth.HealthCheckResponse, error) {
    48  	return &grpchealth.HealthCheckResponse{
    49  		Status: grpchealth.HealthCheckResponse_SERVING,
    50  	}, nil
    51  }
    52  
    53  func (s successServerMock) Watch(_ *grpchealth.HealthCheckRequest, stream grpchealth.Health_WatchServer) error {
    54  	return stream.Send(&grpchealth.HealthCheckResponse{
    55  		Status: grpchealth.HealthCheckResponse_SERVING,
    56  	})
    57  }
    58  
    59  type errorTimeoutServerMock struct {
    60  }
    61  
    62  func (e errorTimeoutServerMock) Check(context.Context, *grpchealth.HealthCheckRequest) (*grpchealth.HealthCheckResponse, error) {
    63  	time.Sleep(time.Second * 4)
    64  	return &grpchealth.HealthCheckResponse{
    65  		Status: grpchealth.HealthCheckResponse_SERVING,
    66  	}, nil
    67  }
    68  
    69  func (e errorTimeoutServerMock) Watch(_ *grpchealth.HealthCheckRequest, stream grpchealth.Health_WatchServer) error {
    70  	time.Sleep(time.Second * 4)
    71  	return stream.Send(&grpchealth.HealthCheckResponse{
    72  		Status: grpchealth.HealthCheckResponse_SERVING,
    73  	})
    74  }
    75  
    76  type errorNotServeServerMock struct {
    77  }
    78  
    79  func (e errorNotServeServerMock) Check(context.Context, *grpchealth.HealthCheckRequest) (*grpchealth.HealthCheckResponse, error) {
    80  	return &grpchealth.HealthCheckResponse{
    81  		Status: grpchealth.HealthCheckResponse_NOT_SERVING,
    82  	}, nil
    83  }
    84  
    85  func (e errorNotServeServerMock) Watch(_ *grpchealth.HealthCheckRequest, stream grpchealth.Health_WatchServer) error {
    86  	return stream.Send(&grpchealth.HealthCheckResponse{
    87  		Status: grpchealth.HealthCheckResponse_NOT_SERVING,
    88  	})
    89  }
    90  
    91  func TestGrpcProber_Probe(t *testing.T) {
    92  	t.Run("Should: failed but return nil error because cant find host", func(t *testing.T) {
    93  		s := New()
    94  		p, o, err := s.Probe("", "", 32, time.Second)
    95  		assert.Equal(t, probe.Failure, p)
    96  		assert.Equal(t, nil, err)
    97  		assert.Equal(t, "timeout: failed to connect service \":32\" within 1s: context deadline exceeded", o)
    98  	})
    99  	t.Run("Should: return nil error because connection closed", func(t *testing.T) {
   100  		s := New()
   101  		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
   102  			fmt.Fprint(w, "res")
   103  		}))
   104  		u := strings.Split(server.URL, ":")
   105  		assert.Equal(t, 3, len(u))
   106  
   107  		port, err := strconv.Atoi(u[2])
   108  		assert.Equal(t, nil, err)
   109  
   110  		// take some time to wait server boot
   111  		time.Sleep(2 * time.Second)
   112  		p, _, err := s.Probe("127.0.0.1", "", port, time.Second)
   113  		assert.Equal(t, probe.Failure, p)
   114  		assert.Equal(t, nil, err)
   115  	})
   116  	t.Run("Should: return nil error because server response not served", func(t *testing.T) {
   117  		s := New()
   118  		lis, _ := net.Listen("tcp", ":0")
   119  		port := lis.Addr().(*net.TCPAddr).Port
   120  		grpcServer := grpc.NewServer()
   121  		defer grpcServer.Stop()
   122  		grpchealth.RegisterHealthServer(grpcServer, &errorNotServeServerMock{})
   123  		go func() {
   124  			_ = grpcServer.Serve(lis)
   125  		}()
   126  		// take some time to wait server boot
   127  		time.Sleep(2 * time.Second)
   128  		p, o, err := s.Probe("0.0.0.0", "", port, time.Second)
   129  		assert.Equal(t, probe.Failure, p)
   130  		assert.Equal(t, nil, err)
   131  		assert.Equal(t, "service unhealthy (responded with \"NOT_SERVING\")", o)
   132  	})
   133  	t.Run("Should: return nil-error because server not response in time", func(t *testing.T) {
   134  		s := New()
   135  		lis, _ := net.Listen("tcp", ":0")
   136  		port := lis.Addr().(*net.TCPAddr).Port
   137  
   138  		grpcServer := grpc.NewServer()
   139  		defer grpcServer.Stop()
   140  		grpchealth.RegisterHealthServer(grpcServer, &errorTimeoutServerMock{})
   141  		go func() {
   142  			_ = grpcServer.Serve(lis)
   143  		}()
   144  		// take some time to wait server boot
   145  		time.Sleep(2 * time.Second)
   146  		p, o, err := s.Probe("0.0.0.0", "", port, time.Second*2)
   147  		assert.Equal(t, probe.Failure, p)
   148  		assert.Equal(t, nil, err)
   149  		assert.Equal(t, "timeout: health rpc did not complete within 2s", o)
   150  
   151  	})
   152  	t.Run("Should: not return error because check was success", func(t *testing.T) {
   153  		s := New()
   154  		lis, _ := net.Listen("tcp", ":0")
   155  		port := lis.Addr().(*net.TCPAddr).Port
   156  
   157  		grpcServer := grpc.NewServer()
   158  		defer grpcServer.Stop()
   159  		grpchealth.RegisterHealthServer(grpcServer, &successServerMock{})
   160  		go func() {
   161  			_ = grpcServer.Serve(lis)
   162  		}()
   163  		// take some time to wait server boot
   164  		time.Sleep(2 * time.Second)
   165  		p, _, err := s.Probe("0.0.0.0", "", port, time.Second*2)
   166  		assert.Equal(t, probe.Success, p)
   167  		assert.Equal(t, nil, err)
   168  	})
   169  	t.Run("Should: not return error because check was success, when listen port is 0", func(t *testing.T) {
   170  		s := New()
   171  		lis, _ := net.Listen("tcp", ":0")
   172  		port := lis.Addr().(*net.TCPAddr).Port
   173  
   174  		grpcServer := grpc.NewServer()
   175  		defer grpcServer.Stop()
   176  		grpchealth.RegisterHealthServer(grpcServer, &successServerMock{})
   177  		go func() {
   178  			_ = grpcServer.Serve(lis)
   179  		}()
   180  		// take some time to wait server boot
   181  		time.Sleep(2 * time.Second)
   182  		p, _, err := s.Probe("0.0.0.0", "", port, time.Second*2)
   183  		assert.Equal(t, probe.Success, p)
   184  		assert.Equal(t, nil, err)
   185  	})
   186  }
   187
View as plain text