FilesExpand file tree

 master

/

config.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

176 lines (140 loc) · 7 KB

 master

/

config.py

Copy path

Top

File metadata and controls

• Code
• Blame

176 lines (140 loc) · 7 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

import re

import os

import socket

import subprocess

import tempfile

import logging

from os.path import expanduser

from six import iteritems

from boto3 import Session

from localstack.constants import DEFAULT_SERVICE_PORTS, LOCALHOST, PATH_USER_REQUEST, DEFAULT_PORT_WEB_UI

# randomly inject faults to Kinesis

KINESIS_ERROR_PROBABILITY = float(os.environ.get('KINESIS_ERROR_PROBABILITY', '').strip() or 0.0)

# randomly inject faults to DynamoDB

DYNAMODB_ERROR_PROBABILITY = float(os.environ.get('DYNAMODB_ERROR_PROBABILITY', '').strip() or 0.0)

# expose services on a specific host internally

HOSTNAME = os.environ.get('HOSTNAME', '').strip() or LOCALHOST

# expose services on a specific host externally

HOSTNAME_EXTERNAL = os.environ.get('HOSTNAME_EXTERNAL', '').strip() or LOCALHOST

# name of the host under which the LocalStack services are available

LOCALSTACK_HOSTNAME = os.environ.get('LOCALSTACK_HOSTNAME', '').strip() or HOSTNAME

# whether to remotely copy the lambda or locally mount a volume

LAMBDA_REMOTE_DOCKER = os.environ.get('LAMBDA_REMOTE_DOCKER', '').lower().strip() in ['true', '1']

# folder for temporary files and data

TMP_FOLDER = os.path.join(tempfile.gettempdir(), 'localstack')

# fix for Mac OS, to be able to mount /var/folders in Docker

if TMP_FOLDER.startswith('/var/folders/') and os.path.exists('/private%s' % TMP_FOLDER):

TMP_FOLDER = '/private%s' % TMP_FOLDER

# temporary folder of the host (required when running in Docker). Fall back to local tmp folder if not set

HOST_TMP_FOLDER = os.environ.get('HOST_TMP_FOLDER', TMP_FOLDER)

# directory for persisting data

DATA_DIR = os.environ.get('DATA_DIR', '').strip()

# whether to use SSL encryption for the services

USE_SSL = os.environ.get('USE_SSL', '').strip() not in ('0', 'false', '')

# default encoding used to convert strings to byte arrays (mainly for Python 3 compatibility)

DEFAULT_ENCODING = 'utf-8'

# path to local Docker UNIX domain socket

DOCKER_SOCK = os.environ.get('DOCKER_SOCK', '').strip() or '/var/run/docker.sock'

# port of Web UI

PORT_WEB_UI = int(os.environ.get('PORT_WEB_UI', '').strip() or DEFAULT_PORT_WEB_UI)

# whether to use Lambda functions in a Docker container

LAMBDA_EXECUTOR = os.environ.get('LAMBDA_EXECUTOR', '').strip()

if not LAMBDA_EXECUTOR:

LAMBDA_EXECUTOR = 'local'

try:

if 'Linux' in subprocess.check_output('uname -a'):

LAMBDA_EXECUTOR = 'docker'

except Exception as e:

pass

# list of environment variable names used for configuration.

# Make sure to keep this in sync with the above!

# Note: do *not* include DATA_DIR in this list, as it is treated separately

CONFIG_ENV_VARS = ['SERVICES', 'HOSTNAME', 'HOSTNAME_EXTERNAL', 'LOCALSTACK_HOSTNAME',

'LAMBDA_EXECUTOR', 'LAMBDA_REMOTE_DOCKER', 'USE_SSL', 'LICENSE_KEY', 'DEBUG',

'KINESIS_ERROR_PROBABILITY', 'DYNAMODB_ERROR_PROBABILITY', 'PORT_WEB_UI']

for key, value in iteritems(DEFAULT_SERVICE_PORTS):

backend_override_var = '%s_BACKEND' % key.upper().replace('-', '_')

if os.environ.get(backend_override_var):

CONFIG_ENV_VARS.append(backend_override_var)

def in_docker():

""" Returns: True if running in a docker container, else False """

if not os.path.exists('/proc/1/cgroup'):

return False

with open('/proc/1/cgroup', 'rt') as ifh:

return 'docker' in ifh.read()

# determine route to Docker host from container

DOCKER_BRIDGE_IP = '172.17.0.1'

try:

DOCKER_HOST_FROM_CONTAINER = socket.gethostbyname('host.docker.internal')

# update LOCALSTACK_HOSTNAME if host.docker.internal is available

if in_docker() and LOCALSTACK_HOSTNAME == DOCKER_BRIDGE_IP:

LOCALSTACK_HOSTNAME = DOCKER_HOST_FROM_CONTAINER

except socket.error:

DOCKER_HOST_FROM_CONTAINER = DOCKER_BRIDGE_IP

# make sure we default to LAMBDA_REMOTE_DOCKER=true if running in Docker

if in_docker() and not os.environ.get('LAMBDA_REMOTE_DOCKER', '').strip():

LAMBDA_REMOTE_DOCKER = True

# local config file path in home directory

CONFIG_FILE_PATH = os.path.join(expanduser("~"), '.localstack')

# create folders

for folder in [DATA_DIR, TMP_FOLDER]:

if folder and not os.path.exists(folder):

try:

os.makedirs(folder)

except Exception as e:

# this can happen due to a race condition when starting

# multiple processes in parallel. Should be safe to ignore

pass

# set variables no_proxy, i.e., run internal service calls directly

no_proxy = ','.join(set((LOCALSTACK_HOSTNAME, HOSTNAME, LOCALHOST, '127.0.0.1', '[::1]')))

if os.environ.get('no_proxy'):

os.environ['no_proxy'] += ',' + no_proxy

elif os.environ.get('NO_PROXY'):

os.environ['NO_PROXY'] += ',' + no_proxy

else:

os.environ['no_proxy'] = no_proxy

# additional CLI commands, can be set by plugins

CLI_COMMANDS = {}

# set of valid regions

VALID_REGIONS = set(Session().get_available_regions('sns'))

def parse_service_ports():

""" Parses the environment variable $SERVICE_PORTS with a comma-separated list of services

and (optional) ports they should run on: 'service1:port1,service2,service3:port3' """

service_ports = os.environ.get('SERVICES', '').strip()

if not service_ports:

return DEFAULT_SERVICE_PORTS

result = {}

for service_port in re.split(r'\s*,\s*', service_ports):

parts = re.split(r'[:=]', service_port)

service = parts[0]

result[service] = int(parts[-1]) if len(parts) > 1 else DEFAULT_SERVICE_PORTS.get(service)

# Fix Elasticsearch port - we have 'es' (AWS ES API) and 'elasticsearch' (actual Elasticsearch API)

if result.get('es') and not result.get('elasticsearch'):

result['elasticsearch'] = DEFAULT_SERVICE_PORTS.get('elasticsearch')

return result

def populate_configs():

global SERVICE_PORTS

SERVICE_PORTS = parse_service_ports()

# define service ports and URLs as environment variables

for key, value in iteritems(DEFAULT_SERVICE_PORTS):

key_upper = key.upper().replace('-', '_')

# define PORT_* variables with actual service ports as per configuration

exec('global PORT_%s; PORT_%s = SERVICE_PORTS.get("%s", 0)' % (key_upper, key_upper, key))

url = 'http%s://%s:%s' % ('s' if USE_SSL else '', LOCALSTACK_HOSTNAME, SERVICE_PORTS.get(key, 0))

# define TEST_*_URL variables with mock service endpoints

exec('global TEST_%s_URL; TEST_%s_URL = "%s"' % (key_upper, key_upper, url))

# expose HOST_*_URL variables as environment variables

os.environ['TEST_%s_URL' % key_upper] = url

# expose LOCALSTACK_HOSTNAME as env. variable

os.environ['LOCALSTACK_HOSTNAME'] = LOCALSTACK_HOSTNAME

def service_port(service_key):

return SERVICE_PORTS.get(service_key, 0)

# initialize config values

populate_configs()

# set log level

if os.environ.get('DEBUG', '').lower() in ('1', 'true'):

logging.getLogger('').setLevel(logging.DEBUG)

logging.getLogger('localstack').setLevel(logging.DEBUG)

# set URL pattern of inbound API gateway

INBOUND_GATEWAY_URL_PATTERN = ('%s/restapis/{api_id}/{stage_name}/%s{path}' %

(TEST_APIGATEWAY_URL, PATH_USER_REQUEST)) # flake8: noqa