return path
+def get_smo_ca_config_path():
+ path = os.environ.get("SMO_CA_CONFIG", "/configs/smoca.crt")
+ return path
+
+
def get_postgres_uri():
host = os.environ.get("DB_HOST", "localhost")
port = 54321 if host == "localhost" else 5432
return events_yaml_name
return "/configs/events.yaml"
-# get k8s host from env:
-
+# get k8s host from env:
def get_k8s_host():
k8s_host = os.environ.get("KUBERNETES_SERVICE_HOST")
if k8s_host is None:
raise Exception('Get k8s host failed.')
return k8s_host
-# get k8s host port from env:
-
+# get k8s host port from env:
def get_k8s_port():
k8s_port = os.environ.get("KUBERNETES_SERVICE_PORT_HTTPS", '443')
return k8s_port
-# token review url
-
+# token review url
def get_review_url():
try:
api = '/apis/authentication.k8s.io/v1/tokenreviews'
except Exception:
raise Exception('Get k8s review url failed')
-# get reviewer token
-
+# get reviewer token
def get_reviewer_token():
# token path default is below.
token_path = '/var/run/secrets/kubernetes.io/serviceaccount/token'
def get_auth_provider():
return 'k8s'
+
+
+def get_dms_support_profiles():
+ profiles = config.conf.API.DMS_SUPPORT_PROFILES
+ if profiles is None or profiles == '':
+ profiles = []
+ elif "[" in profiles and "]" in profiles:
+ profiles = profiles.replace("'", "").replace(
+ '"', "").replace('[', "").replace(']', "")
+ profiles = profiles.split(',')
+ if 'native_k8sapi' not in profiles:
+ profiles.append('native_k8sapi')
+ return profiles