GG-32415: Fix timezone mess when using SQL (#21)

Author: isapego

pygridgain/connection/__init__.py

@@ -23,12 +23,13 @@
 import socket
 from threading import RLock
 from typing import Union
+from tzlocal import get_localzone
 
 from pygridgain.constants import *
 from pygridgain.exceptions import (
     HandshakeError, ParameterError, SocketError, connection_errors,
 )
-from pygridgain.datatypes import Byte, Int, Short, String, UUIDObject
+from pygridgain.datatypes import Byte, ByteArrayObject, Int, Short, String, UUIDObject
 from pygridgain.datatypes.internal import Struct
 from pygridgain.utils import DaemonicTimer
 
@@ -204,6 +205,11 @@ def read_response(self) -> Union[dict, OrderedDict]:
                     ('version_patch', Short),
                     ('message', String),
                 ])
+            elif self.get_protocol_version() >= (1, 7, 0):
+                response_end = Struct([
+                    ('features', ByteArrayObject),
+                    ('node_uuid', UUIDObject),
+                ])
             elif self.get_protocol_version() >= (1, 4, 0):
                 response_end = Struct([
                     ('node_uuid', UUIDObject),
@@ -276,10 +282,13 @@ def _connect_version(
 
         protocol_version = self.client.protocol_version
 
+        timezone = get_localzone().tzname(None)
+
         hs_request = HandshakeRequest(
-            protocol_version,
-            self.username,
-            self.password
+            protocol_version=protocol_version,
+            username=self.username,
+            password=self.password,
+            timezone=timezone,
         )
 
         with BinaryStream(self) as stream:

pygridgain/connection/handshake.py

@@ -15,22 +15,25 @@
 #
 from typing import Optional, Tuple
 
-from pygridgain.datatypes import Byte, Int, Short, String
+from pygridgain.datatypes import Byte, ByteArrayObject, Int, MapObject, Short, String
 from pygridgain.datatypes.internal import Struct
 
 OP_HANDSHAKE = 1
 
+USER_ATTR_TIMEZONE = 'client.timezone'
+
 
 class HandshakeRequest:
     """ Handshake request. """
     handshake_struct = None
     username = None
     password = None
     protocol_version = None
+    timezone = None
 
     def __init__(
         self, protocol_version: Tuple[int, int, int],
-        username: Optional[str] = None, password: Optional[str] = None
+        username: Optional[str] = None, password: Optional[str] = None, timezone: str = None,
     ):
         fields = [
             ('length', Int),
@@ -41,6 +44,15 @@ def __init__(
             ('client_code', Byte),
         ]
         self.protocol_version = protocol_version
+        self.timezone = timezone
+        if protocol_version >= (1, 7, 0):
+            fields.extend([
+                ('features', ByteArrayObject),
+            ])
+        if protocol_version >= (1, 7, 1):
+            fields.extend([
+                ('user_attributes', MapObject),
+            ])
         if username and password:
             self.username = username
             self.password = password
@@ -59,6 +71,21 @@ def from_python(self, stream):
             'version_patch': self.protocol_version[2],
             'client_code': 2,  # fixed value defined by protocol
         }
+        if self.protocol_version >= (1, 7, 0):
+            handshake_data.update({
+                'features': None,
+            })
+            handshake_data['length'] += 1
+        if self.protocol_version >= (1, 7, 1):
+            user_attributes = (
+                MapObject.HASH_MAP, {
+                    USER_ATTR_TIMEZONE: self.timezone
+                })
+
+            handshake_data.update({
+                'user_attributes': user_attributes,
+            })
+            handshake_data['length'] += 6 + 5 + len(USER_ATTR_TIMEZONE) + 5 + len(self.timezone)
         if self.username and self.password:
             handshake_data.update({
                 'username': self.username,
@@ -69,5 +96,4 @@ def from_python(self, stream):
                 len(self.username),
                 len(self.password),
             ])
-
         self.handshake_struct.from_python(stream, handshake_data)

pygridgain/constants.py

@@ -30,14 +30,18 @@
 ]
 
 PROTOCOLS = {
+    (1, 7, 1),
+    (1, 7, 0),
+    (1, 6, 0),
+    (1, 5, 0),
     (1, 4, 0),
     (1, 3, 0),
     (1, 2, 0),
 }
 
 PROTOCOL_VERSION_MAJOR = 1
-PROTOCOL_VERSION_MINOR = 4
-PROTOCOL_VERSION_PATCH = 0
+PROTOCOL_VERSION_MINOR = 7
+PROTOCOL_VERSION_PATCH = 1
 
 MAX_LONG = 9223372036854775807
 MIN_LONG = -9223372036854775808

requirements/install.txt

@@ -2,3 +2,4 @@
 
 typing==3.6.6; python_version<'3.5'
 attrs==18.1.0
+tzlocal==2.1

tests/config/ignite-config.xml.jinja2

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 
 <!--
- Copyright 2019 GridGain Systems, Inc. and Contributors.
+ Copyright 2021 GridGain Systems, Inc. and Contributors.
 
  Licensed under the GridGain Community Edition License (the "License");
  you may not use this file except in compliance with the License.
@@ -59,12 +59,12 @@
         <property name="discoverySpi">
             <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
                 <property name="localAddress" value="127.0.0.1"/>
-                <property name="localPort" value="48500"/>
+                <property name="localPort" value="{{ discovery_port }}"/>
                 <property name="ipFinder">
                     <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">
                         <property name="addresses">
                             <list>
-                                <value>127.0.0.1:48500..48510</value>
+                                <value>127.0.0.1:{{ discovery_port }}..{{ discovery_port_range }}</value>
                             </list>
                         </property>
                     </bean>
@@ -76,7 +76,7 @@
         <property name="communicationSpi">
             <bean class="org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi">
                 <property name="localAddress" value="127.0.0.1"/>
-                <property name="localPort" value="48100"/>
+                <property name="localPort" value="{{ communication_port }}"/>
             </bean>
         </property>
 

tests/conftest.py

@@ -82,8 +82,8 @@ def server3(request):
 
 @pytest.fixture(scope='module')
 def start_ignite_server(use_ssl):
-    def start(idx=1):
-        return _start_ignite(idx, use_ssl=use_ssl)
+    def start(idx=1, cluster_idx=1, jvm_opts=''):
+        return _start_ignite(idx, use_ssl=use_ssl, cluster_idx=cluster_idx, jvm_opts=jvm_opts)
 
     return start
 

tests/test_sql.py

@@ -13,14 +13,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+from datetime import datetime
+
+import pytest
+
 from pygridgain.api import (
     sql_fields, sql_fields_cursor_get_page,
-    cache_get_or_create, sql, sql_cursor_get_page,
+    sql, sql_cursor_get_page,
     cache_get_configuration,
 )
 from pygridgain.datatypes.prop_codes import *
 from pygridgain.utils import entity_id
 from pygridgain.binary import unwrap_binary
+from tests.util import kill_process_tree
 
 initial_data = [
         ('John', 'Doe', 5),
@@ -187,3 +192,27 @@ def test_long_multipage_query(client):
             assert value == field_number * page[0]
 
     client.sql(drop_query)
+
+
+@pytest.mark.parametrize('timezone', ['UTC', 'GMT+5', 'GMT-3'])
+def test_server_in_different_timezone(start_ignite_server, start_client, timezone):
+    server_id = 10
+    server = start_ignite_server(idx=server_id, cluster_idx=2, jvm_opts=f'-Duser.timezone={timezone}')
+    try:
+        client = start_client()
+        client.connect('127.0.0.1', 10800 + server_id)
+
+        client.get_or_create_cache('PUBLIC')
+        client.sql('create table test(key int primary key, time datetime)')
+
+        current_time = datetime(year=2020, month=2, day=12, hour=12, minute=32, second=55)
+        client.sql(f"insert into test (key, time) VALUES (1, '{current_time}')")
+
+        page = client.sql('SELECT time FROM test')
+        received = next(page)[0][0]
+
+        assert current_time == received
+
+        client.close()
+    finally:
+        kill_process_tree(server.pid)

tests/util.py

@@ -76,7 +76,7 @@ def get_ignite_config_path(use_ssl=False):
     if use_ssl:
         file_name = "ignite-config-ssl.xml"
     else:
-        file_name = "ignite-config.xml.jinja2"
+        file_name = "ignite-default.xml.jinja2"
 
     return os.path.join(get_test_dir(), "config", file_name)
 
@@ -113,18 +113,30 @@ def create_config_file(tpl_name, file_name, **kwargs):
         f.write(template.render(**kwargs))
 
 
-def _start_ignite(idx=1, debug=False, use_ssl=False):
+def _start_ignite(idx=1, debug=False, use_ssl=False, cluster_idx=1, jvm_opts=''):
     clear_logs(idx)
 
     runner = get_ignite_runner()
 
     env = os.environ.copy()
 
     if debug:
-        env["JVM_OPTS"] = "-Djava.net.preferIPv4Stack=true -Xdebug -Xnoagent -Djava.compiler=NONE " \
-                          "-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005 "
-
-    params = {'ignite_instance_idx': str(idx), 'ignite_client_port': 10800 + idx, 'use_ssl': use_ssl}
+        env["JVM_OPTS"] = env.get("JVM_OPTS", '') + \
+                          "-Djava.net.preferIPv4Stack=true -Xdebug -Xnoagent -Djava.compiler=NONE " \
+                          "-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address={5005} "
+
+    if jvm_opts:
+        env["JVM_OPTS"] = env.get("JVM_OPTS", '') + jvm_opts
+
+    port_offset = (cluster_idx - 1) * 10
+    params = {
+        'ignite_instance_idx': str(idx),
+        'ignite_client_port': 10800 + idx,
+        'use_ssl': use_ssl,
+        'discovery_port': 48500 + port_offset,
+        'discovery_port_range': 48510 + port_offset,
+        'communication_port': 48100 + port_offset,
+    }
 
     create_config_file('log4j.xml.jinja2', f'log4j-{idx}.xml', **params)
     create_config_file('ignite-config.xml.jinja2', f'ignite-config-{idx}.xml', **params)