Use Elasticsearch readiness port

pkg/controller/elasticsearch/services/services.go

@@ -227,6 +227,8 @@ func (u *urlProvider) HasEndpoints() bool {
  return err == nil && len(k8s.RunningPods(pods)) > 0
 }
 
+// NewElasticsearchURLProvider returns a client.URLProvider that dynamically tries to find Pod URLs among the
+// currently running Pods. Preferring ready Pods over running ones.
 func NewElasticsearchURLProvider(es esv1.Elasticsearch, client k8s.Client) client.URLProvider {
  return &urlProvider{
  pods: func() ([]corev1.Pod, error) {

pkg/controller/elasticsearch/services/services_test.go

@@ -403,7 +403,7 @@ func TestNewElasticsearchURLProvider(t *testing.T) {
  wantErr bool
  }{
  {
- name: "cache failures are swallowed but logged",
+ name: "cache failures are returned to the caller",
  args: args{
  es: mkElasticsearch(commonv1.HTTPConfig{}),
  client: k8s.NewFailingClient(errors.New("boom")),

pkg/utils/k8s/k8sutils.go

@@ -74,6 +74,7 @@ func IsPodReady(pod corev1.Pod) bool {
  return conditionsTrue == 2
 }
 
+// IsPodRunning returns true if the Pod is in phase running and not terminating.
 func IsPodRunning(pod corev1.Pod) bool {
  return pod.DeletionTimestamp.IsZero() && pod.Status.Phase == corev1.PodRunning
 }
@@ -90,6 +91,7 @@ func TerminatingPods(pods []corev1.Pod) []corev1.Pod {
  return terminating
 }
 
+// RunningPods filters pods for Pods are in running (and not terminating).
 func RunningPods(pods []corev1.Pod) []corev1.Pod {
  var running []corev1.Pod
  for _, p := range pods {

test/e2e/es/forced_upgrade_test.go

@@ -103,8 +103,7 @@ func TestForceUpgradePendingPodsInOneStatefulSet(t *testing.T) {
  if err != nil {
  return err
  }
- running := k8s.RunningPods(pods)
- if len(running) == 0 {
+ if len(k8s.RunningPods(pods)) == 0 {
  return errors.New("Elasticsearch does not have any running Pods")
  }
  return nil