Merge pull request #96 from zalando/bugfix/haproxy_errors

Handle haproxy requests. Improve failover status code.

haproxy.cfg

@@ -15,8 +15,7 @@ frontend ft_postgresql
 	default_backend bk_db
 
 backend bk_db
-	option httpchk GET /master
-    http-check expect string '"master"'
+	option httpchk
 
     server postgresql_127.0.0.1_5432 127.0.0.1:5432 maxconn 100 check port 8008
     server postgresql_127.0.0.1_5433 127.0.0.1:5433 maxconn 100 check port 8009

patroni/api.py

@@ -3,6 +3,7 @@ import fcntl
 import json
 import logging
 import psycopg2
+import socket
 import time
 
 from patroni.exceptions import PostgresConnectionException
@@ -37,12 +38,24 @@ class RestApiHandler(BaseHTTPRequestHandler):
         self.end_headers()
         self.wfile.write(body.encode('utf-8'))
 
+    def finish(self, *args, **kwargs):
+        try:
+            if not self.wfile.closed:
+                self.wfile.flush()
+                self.wfile.close()
+        except socket.error:
+            pass
+        self.rfile.close()
+
     def check_auth_header(self):
         auth_header = self.headers.get('Authorization')
         status = self.server.check_auth_header(auth_header)
         return not status or self.send_auth_request(status)
 
-    def do_GET(self):
+    def do_OPTIONS(self):
+        self.do_GET(options=True)
+
+    def do_GET(self, options=False):
         """Default method for processing all GET requests which can not be routed to other methods"""
 
         path = '/master' if self.path == '/' else self.path
@@ -70,9 +83,10 @@ class RestApiHandler(BaseHTTPRequestHandler):
             status_code = 503
 
         self.send_response(status_code)
-        self.send_header('Content-Type', 'application/json')
-        self.end_headers()
-        self.wfile.write(json.dumps(response).encode('utf-8'))
+        if not options:
+            self.send_header('Content-Type', 'application/json')
+            self.end_headers()
+            self.wfile.write(json.dumps(response).encode('utf-8'))
 
     def do_GET_patroni(self):
         response = self.get_postgresql_status(True)
@@ -190,6 +204,12 @@ class RestApiHandler(BaseHTTPRequestHandler):
                 self.command = mname
         return ret
 
+    def handle_one_request(self):
+        try:
+            BaseHTTPRequestHandler.handle_one_request(self)
+        except socket.error:
+            pass
+
     def query(self, sql, *params, **kwargs):
         if not kwargs.get('retry', False):
             return self.server.query(sql, *params)

patroni/ha.py

@@ -273,8 +273,9 @@ class Ha:
         if self.is_healthiest_node():
             if self.acquire_lock():
                 if self.cluster.failover:
-                    logger.info('Cleanning up failover key after acquiring leader lock...')
+                    logger.info('Cleaning up failover key after acquiring leader lock...')
                     self.dcs.manual_failover('', '')
+                self.dcs.get_cluster()
                 return self.enforce_master_role('acquired session lock as a leader',
                                                 'promoted self to leader by acquiring session lock')
             else:

tests/test_api.py

@@ -6,6 +6,8 @@ from patroni.api import RestApiHandler, RestApiServer
 from patroni.dcs import Member
 from six import BytesIO as IO
 from six.moves import BaseHTTPServer
+from six.moves.BaseHTTPServer import BaseHTTPRequestHandler
+import socket
 from test_postgresql import psycopg2_connect, MockCursor
 
 
@@ -90,6 +92,20 @@ class TestRestApiHandler(unittest.TestCase):
             MockRestApiServer(RestApiHandler, b'GET /master')
         MockRestApiServer(RestApiHandler, b'GET /master')
 
+    def test_do_OPTIONS(self):
+        MockRestApiServer(RestApiHandler, b'OPTIONS / HTTP/1.0')
+
+        with patch.object(BaseHTTPRequestHandler, 'handle_one_request') as mock_handle_request:
+            mock_handle_request.side_effect = socket.error("foo")
+            MockRestApiServer(RestApiHandler, b'OPTIONS / HTTP/1.0')
+
+        # make sure socket.error gets propagated via wfile object in finalize()
+        with patch.object(MockRequest, 'makefile') as makefile:
+            makefile.return_value.closed = False
+            makefile.return_value.readline.side_effect = lambda x: b"foo"
+            makefile.return_value.flush = Mock(side_effect=socket.error("foo"))
+            MockRestApiServer(RestApiHandler, b'OPTIONS / HTTP/1.0')
+
     def test_do_GET_patroni(self):
         MockRestApiServer(RestApiHandler, b'GET /patroni')