apache-ignite

1
# Licensed to the Apache Software Foundation (ASF) under one or more
2
# contributor license agreements.  See the NOTICE file distributed with
3
# this work for additional information regarding copyright ownership.
4
# The ASF licenses this file to You under the Apache License, Version 2.0
5
# (the "License"); you may not use this file except in compliance with
6
# the License.  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
Module contains classes and utility methods to create discovery configuration for ignite nodes.
18
"""
19

20
from abc import ABCMeta, abstractmethod
21

22
from ignitetest.services.utils.ignite_aware import IgniteAwareService
23
from ignitetest.services.zk.zookeeper import ZookeeperService
24

25

26
class DiscoverySpi(metaclass=ABCMeta):
27
    """
28
    Abstract class for DiscoverySpi.
29
    """
30
    @property
31
    @abstractmethod
32
    def type(self):
33
        """
34
        Type of DiscoverySPI.
35
        """
36

37
    @abstractmethod
38
    def prepare_on_start(self, **kwargs):
39
        """
40
        Call if update before start is needed.
41
        """
42

43

44
class ZookeeperDiscoverySpi(DiscoverySpi):
45
    """
46
    ZookeeperDiscoverySpi.
47
    """
48
    def __init__(self, zoo_service, root_path):
49
        self.connection_string = zoo_service.connection_string()
50
        self.port = zoo_service.settings.client_port
51
        self.root_path = root_path
52
        self.session_timeout = zoo_service.settings.min_session_timeout
53

54
    @property
55
    def type(self):
56
        return "ZOOKEEPER"
57

58
    def prepare_on_start(self, **kwargs):
59
        pass
60

61

62
class TcpDiscoveryIpFinder(metaclass=ABCMeta):
63
    """
64
    Abstract class for TcpDiscoveryIpFinder.
65
    """
66
    @property
67
    @abstractmethod
68
    def type(self):
69
        """
70
        Type of TcpDiscoveryIpFinder.
71
        """
72

73
    @abstractmethod
74
    def prepare_on_start(self, **kwargs):
75
        """
76
        Call if update before start is needed.
77
        """
78

79

80
class TcpDiscoveryVmIpFinder(TcpDiscoveryIpFinder):
81
    """
82
    IpFinder with static ips, obtained from cluster nodes.
83
    """
84
    def __init__(self, nodes=None):
85
        self.addresses = TcpDiscoveryVmIpFinder.__get_addresses(nodes) if nodes else None
86

87
    @property
88
    def type(self):
89
        return 'VM'
90

91
    def prepare_on_start(self, **kwargs):
92
        if not self.addresses:
93
            cluster = kwargs.get('cluster')
94
            self.addresses = TcpDiscoveryVmIpFinder.__get_addresses(cluster.nodes)
95

96
    @staticmethod
97
    def __get_addresses(nodes):
98
        return [node.account.externally_routable_ip for node in nodes]
99

100

101
class TcpDiscoverySpi(DiscoverySpi):
102
    """
103
    TcpDiscoverySpi.
104
    """
105
    def __init__(self, ip_finder=TcpDiscoveryVmIpFinder(), port=47500, port_range=100, local_address=None):
106
        self.ip_finder = ip_finder
107
        self.port = port
108
        self.port_range = port_range
109
        self.local_address = local_address
110

111
    @property
112
    def type(self):
113
        return 'TCP'
114

115
    def prepare_on_start(self, **kwargs):
116
        self.ip_finder.prepare_on_start(**kwargs)
117

118

119
def from_ignite_cluster(cluster, subset=None):
120
    """
121
    Form TcpDiscoverySpi from cluster or its subset.
122
    :param cluster: IgniteService cluster
123
    :param subset: slice object (optional).
124
    :return: TcpDiscoverySpi with static ip addresses.
125
    """
126
    assert isinstance(cluster, IgniteAwareService)
127

128
    if subset:
129
        assert isinstance(subset, slice)
130
        nodes = cluster.nodes[subset]
131
    else:
132
        nodes = cluster.nodes
133

134
    return TcpDiscoverySpi(ip_finder=TcpDiscoveryVmIpFinder(nodes))
135

136

137
def from_zookeeper_cluster(cluster, root_path="/apacheIgnite"):
138
    """
139
    Form ZookeeperDiscoverySpi from zookeeper service cluster.
140
    :param cluster: ZookeeperService cluster.
141
    :param root_path: root ZNode path.
142
    :return: ZookeeperDiscoverySpi.
143
    """
144
    assert isinstance(cluster, ZookeeperService)
145

146
    return ZookeeperDiscoverySpi(cluster, root_path=root_path)
147