K8s) host명으로 내부 데이터 접근하기_spark,boto3,k8s

IP로 데이터에 접근하다보니, 불안해서 host명으로 수정했어요
k8s에서 작업하시는 분들께 도움이 되었으면 합니다 :) 

요약

1. 문제 : IP로 데이터 접근하니, 보안 리스크가 생김
2. 상황 : spark image, jupyterhub, airflow 에 적용가능함
3. 조치 : IP를 host명으로 대체함
4. 결론 : 보안 리스크 줄이고, cluster 관리가 편해짐

---

설명

1. 문제

data에 IP로 접근해서 생길 수 있는 보안 이슈 발견

k8s cluster를 여러개 관리하다보니까, IP가 이제 꼬이기 시작했어요.

보안 이슈도 있었지만, 곧 신규 cluster 구축이 예정되어있어서 미리 변경하고 싶었어요

제가 편해야 시스템도 안정적으로 변하니까요ㅎ

2. 상황

일단 3가지 먼저, spark image/ jupyterhub/ airflow

모든 것을 한번에 바꾸면 좋겠지만, 제가 관리하고 있는 어플리케이션들에 먼저 적용해보기로 했어요

(k8s cluster는 on-prem이기 때문에 cloud에서 제공하는 좋은 기능들은 사용하지 못해요)

• spark image
  1. image 내부 pyspark
• jupterhub
  1. boto3
  2. s3fs
  3. pyspark
• airflow
  1. connection (remote-logging)

---

3. 조치

3-1. spark image

아래처럼 단순한 sparkSession 로직이 있다고 가정할게요

import pyspark
from pyspark.sql import SparkSession

spark = SparkSession \
        .builder.appName(app_name) \
        .config("spark.hadoop.fs.s3a.path.style.access", "true") \
        .config("spark.hadoop.fs.s3a.access.key", aws_access_key_id) \
        .config("spark.hadoop.fs.s3a.secret.key", aws_secret_access_key) \
        .config("spark.hadoop.fs.s3a.endpoint", endpoint_url) \
        .getOrCreate()

end_point를 ip에서 host명으로 변경해줫어요.

endpoint_url = "http://rook-ceph-rgw-***.rook-ceph.svc.cluster.local:80"

아래 설정값을 넣지 않으면 무한 로딩이 걸릴 수 있으니 주의해주세요

.config("spark.hadoop.fs.s3a.path.style.access", "true") \

변경한 덕분에 ip가 변경되야할 때마다, image를 빌드하지 않아도 되서 뿌듯하네요

 

참고 

https://gist.github.com/tobilg/e03dbc474ba976b9f235

---

3-2. jupterhub

3-2-1. boto3

는 그냥 변경만해서 사용하면 문제 없어요

import boto3
from botocore.client import Config

boto3.resource('s3',
    endpoint_url='http://rook-ceph-rgw-my-store.rook-ceph.svc.cluster.local:80',
    aws_access_key_id='access-key',
    aws_secret_access_key='secret-key',
    config=Config(signature_version='s3v4')
)

3-2-2. s3fs

내부적으로는 boto 사용하니까 동일해요

import s3fs

s3fs.S3FileSystem(
            anon=False,
            use_ssl=False,
            client_kwargs={
                "region_name": "",
                "endpoint_url": endpoint_url,
                "aws_access_key_id": aws_access_key_id,
                "aws_secret_access_key": aws_secret_access_key,
                "verify": False,
            }
        )

3-2-3. pyspark

endopoint는 위의 사례들과 동일하게 host명으로 바꾸면 되는데 master를 바꿔줘야하는게 좀 특이해요.

jupyterhub 안의 terminal에서 host명을 찾을 수 있어요

jovyan@jupyter-manager:~$ export | grep KUBERNETES_PORT_443_TCP
declare -x KUBERNETES_PORT_443_TCP="tcp://10.**.*.1:***"
declare -x KUBERNETES_PORT_443_TCP_ADDR="10.**.*.1"
declare -x KUBERNETES_PORT_443_TCP_PORT="***"
declare -x KUBERNETES_PORT_443_TCP_PROTO="tcp"

.master에 들어갈거를 만들었어요

$ f"k8s://https://{os.environ['KUBERNETES_PORT_443_TCP_ADDR']}:{os.environ['KUBERNETES_PORT_443_TCP_PORT']}"
'k8s://https://10.**.**.1:**$'

아래 설정값도 잊지 않고 넣어줘야해요

.config("spark.hadoop.fs.s3a.path.style.access", "true") \

완성된 코드는 아래와 같아요.

import os, posixpath, socket
import pyspark
from pyspark.sql import SparkSession


spark = (
    SparkSession.builder.appName("hgkim-spark")
        .config("spark.kryoserializer.buffer.max", "1024m")
        .config("spark.sql.sources.partitionOverwriteMode", "dynamic")
        .master(f"k8s://https://{os.environ['KUBERNETES_PORT_443_TCP_ADDR']}:{os.environ['KUBERNETES_PORT_443_TCP_PORT']}")
        .config("spark.kubernetes.container.image", "hgkim/library/spark-3.2.0-base:1.2.0-20211209")
        .config("spark.kubernetes.namespace", "spark_test")
        .config("spark.kubernetes.authenticate.driver.serviceAccountName", "jupyter_test")
        .config("spark.driver.port", "2222")
        .config("spark.driver.blockManager.port", "7777")
        .config("spark.driver.host", socket.gethostbyname(socket.gethostname()))
        .config("spark.driver.bindAddress", "0.0.0.0")
        .config("spark.driver.memory", "8g")
        .config("spark.executor.instances", 3)
        .config("spark.executor.memory", "5g")
        .config("spark.executor.cores", 3)
        .config("spark.dynamicAllocation.enabled", "false")
        .getOrCreate()
)

spark.sparkContext._jsc.hadoopConfiguration().set('fs.s3a.access.key', access_key) #변수
spark.sparkContext._jsc.hadoopConfiguration().set('fs.s3a.secret.key', secret_key) #변수
spark.sparkContext._jsc.hadoopConfiguration().set('fs.s3a.endpoint', endpoint_url) #변수
spark.sparkContext._jsc.hadoopConfiguration().set('fs.s3a.path.style.access', "true") # 필수
spark.sparkContext._jsc.hadoopConfiguration().set('fs.s3a.connection.ssl.enabled', "false")
spark.sparkContext._jsc.hadoopConfiguration().set('fs.s3a.impl', "org.apache.hadoop.fs.s3a.S3AFileSystem")
spark.sparkContext._jsc.hadoopConfiguration().set('fs.s3a.aws.credentials.provider', "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider")
spark.sparkContext._jsc.hadoopConfiguration().set('com.amazonaws.services.s3.enableV4', "true")

spark

잘 동작하네요 ㅎ

$ spark
SparkSession - in-memory
SparkContext
Spark UI
Version  v3.2.0
Master  k8s://https://10.**.**.**:443
AppName  hgkim-spark

$ df = spark.read.format("parquet").load("s3a://hgkim/test.parquet")
$ df.count()
47

참고

https://towardsdatascience.com/jupyter-notebook-spark-on-kubernetes-880af7e06351

Jupyter Notebook & Spark on Kubernetes

3-3. airflow

마지막으로 airflow에요

3-3-1. connection (remote-logging)

helm airflow는 아래와 같은 설정으로 로깅을 해요

- name: "AIRFLOW__CORE__REMOTE_LOGGING"
  value: "True"
- name: "AIRFLOW__CORE__REMOTE_LOG_CONN_ID"
  value: "CephObjectConn"
- name: "AIRFLOW__CORE__REMOTE_BASE_LOG_FOLDER"
  value: "s3://hgkim/airflow/logs"

CephObjectConn의 ip를 host명으로 바꾸어도 잘되네요

{
"aws_access_key_id" : "****" , 
"aws_secret_access_key": "****" ,
"host":"http://rook-ceph-rgw-***.rook-ceph.svc.cluster.local:80"
}

4. 결론

저의 편안함을 위해 host명으로 변경하니 마음이 편하네요.

이제 image 빌드, 어플리케이션 설치 할때마다 ip를 입력할 필요가 없고

사용자에게도 ip가 아닌 host명만 알려주면 되니 편안해요

host명은 k8s cluster 내부에서만 동작하니까, 다른 cluster를 갈 필요도 없고요

(개발계에서 운영계 데이터 당겨오다가, 실수할 일 없음)

 

아무도 알아주지 않는 혼자만의 만족이지만, 만족스럽습니다

 

끗