added function(send)

python/fastapi/app_deployment.yaml

@@ -40,7 +40,7 @@ metadata:
   name: python-fastapi-deployment
   namespace: k8s-edu-ondemand-hashtag
 spec:
-  replicas: 1
+  replicas: 2
   selector:
     matchLabels:
       app: python-fastapi-pod

python/fastapi/models.py

@@ -0,0 +1,12 @@
+from typing import List, Optional
+from pydantic import BaseModel, HttpUrl
+
+class Result(BaseModel):
+    hashtag: Optional[str]
+    caption: Optional[str]
+    like_count: Optional[int]
+    comment_count: Optional[int]
+    related_tags: List[str] = []
+    shortcode: Optional[str]
+    url: Optional[HttpUrl]
+    timestamp: Optional[int]

python/fastapi/server.py

@@ -3,11 +3,13 @@
 import hashlib
 from datetime import datetime, timedelta
 from loguru import logger as loguru_logger
-from fastapi import FastAPI, HTTPException
+from fastapi import FastAPI, HTTPException, Body
 import uvicorn
 import requests
 from starlette.responses import HTMLResponse
-from kubernetes import client, watch
+from kubernetes import client
+from models import Result
+import csv
 import ssl
 
 ssl._create_default_https_context = ssl._create_unverified_context
@@ -27,7 +29,7 @@
 config.verify_ssl = True
 
 batch_v1 = client.BatchV1Api(api_client=client.ApiClient(config))
-
+core_v1 = client.CoreV1Api(api_client=client.ApiClient(config))
 app = FastAPI()
 @app.get("/health_check")
 async def health_check():
@@ -73,6 +75,8 @@ def k8s_trigger_job(hashtag):
     WHATAP_APP_NAME = f"HASHTAG_SEARCH_{hashtag}"
     WHATAP_LOGGING_ENABLED = os.getenv("WHATAP_LOGGING_ENABLED")
 
+    service = core_v1.read_namespaced_service(name="python-fastapi-service", namespace="k8s-edu-ondemand-hashtag")
+    NODE_PORT = service.node_port
     NODE_IP = client.V1EnvVarSource(field_ref=client.V1ObjectFieldSelector(field_path="status.hostIP"))
     NODE_NAME = client.V1EnvVarSource(field_ref=client.V1ObjectFieldSelector(field_path="spec.nodeName"))
     POD_NAME = client.V1EnvVarSource(field_ref=client.V1ObjectFieldSelector(field_path="metadata.name"))
@@ -88,6 +92,7 @@ def create_job_object():
                 client.V1EnvVar(name="HASHTAG", value=hashtag),
                 client.V1EnvVar(name="WHATAP_APP_NAME", value=WHATAP_APP_NAME),
                 client.V1EnvVar(name="NODE_IP", value_from=NODE_IP),
+                client.V1EnvVar(name="NODE_PORT", value_from=NODE_PORT),
                 client.V1EnvVar(name="NODE_NAME", value_from=NODE_NAME),
                 client.V1EnvVar(name="POD_NAME", value_from=POD_NAME)
                 ]
@@ -105,6 +110,7 @@ def create_job_object():
             metadata=client.V1ObjectMeta(name=JOB_NAME),
             spec=spec)
         return job
+
     def create_job(api_instance, job):
         api_response = api_instance.create_namespaced_job(
             body=job,
@@ -122,6 +128,42 @@ def delete_job(api_instance):
     job = create_job_object()
     create_job(api_instance=batch_v1, job=job)
 
+@app.get("/k8s/job/{hashtag}")
+def k8s_get_data(hashtag):
+    if os.path.isfile(f"data/{hashtag}.csv"):
+        mydict = {}
+        with open(file=f"data/{hashtag}.csv", mode="r") as f:
+            reader = csv.reader(f)
+            attrs = []
+            result = []
+            for row in reader:
+                if len(attrs) < 1:
+                    attrs = row
+                    continue
+                temp_row_dict = dict()
+                for key, value in zip(attrs, row):
+                    temp_row_dict[key] = value
+                result.append(temp_row_dict)
+        return result
+    else:
+        return {"message": "no data"}
+
+@app.post("/k8s/save/{hashtag}")
+def k8s_save_data(hashtag, patch: Result=Body(), include_in_schema=False):
+    ### 파일에 performance 데이터 저장하기
+    with open(file=f"data/{hashtag}.csv", mode="a+") as f:
+        f.seek(0)
+        text_append = f"{patch.hashtag}\t{patch.caption}\t{patch.like_count}\t{patch.comment_count}\t{patch.related_tags}\t{patch.shortcode}\t{patch.url}\t{patch.timestamp}"
+
+        if f.read():
+            ### 내용이 존재하면 받아온값을 추가해준다.
+            f.write(f"\n{text_append}")
+
+        else:
+            ### 내용이 존재하지 않으면 칼럼을 먼저 추가하고 내용을 추가한다.
+            f.write(f"hashtag\tcaption\tlike_count\tcomment_count\trelated_tags\tshortcode\turl\ttimestamp")
+            f.write(f"\n{text_append}")
+    return {"hashtag": hashtag}
 
 if __name__ == "__main__":
     uvicorn.run(app="server:app", host="0.0.0.0", port=8000, reload=True)

python/method_profile/services.py

@@ -1,14 +1,21 @@
+import os
 from engine import InstagramScraper
 from extractor import Extractor
 from whatap import method_profiling
 import logging
+import requests
+
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
 stream_handler = logging.StreamHandler()
 stream_handler.setFormatter(formatter)
 logger.addHandler(stream_handler)
 
+
+
+NODE_IP = os.getenv("NODE_IP")
+NODE_PORT = os.getenv("NODE_PORT")
 @method_profiling
 def get_top_posts(hashtag):
 
@@ -24,6 +31,7 @@ def get_top_posts(hashtag):
     logger.info(f"파싱 완료")
 
     logger.info(f"데이터 추가 시작")
+
     for post in data.top_posts:
         with open(file=f"data.csv", mode="a+") as f:
             f.seek(0)
@@ -32,13 +40,26 @@ def get_top_posts(hashtag):
             if f.read():
                 ### 내용이 존재하면 받아온값을 추가해준다.
                 f.write(f"\n{text_append}")
-
             else:
                 ### 내용이 존재하지 않으면 칼럼을 먼저 추가하고 내용을 추가한다.
                 f.write(f"hashtag\tcaption\tlike_count\tcomment_count\trelated_tags\tshortcode\turl\ttimestamp")
                 f.write(f"\n{text_append}")
+
+        send_data = {"hashtag": data.name, "caption": post.caption, "like_count": post.like_count,
+                     "comment_count": post.comment_count, "shortcode": post.shortcode, "url": post.url,
+                     "timestamp": post.taken_at_timestamp}
+        try:
+            requests.post(url=f"https://{NODE_IP}:{NODE_PORT}/k8s/save/{hashtag}", json=send_data)
+        except Exception as e:
+            logger.error(e)
+
     logger.info(f"데이터 추가 완료")
     logger.info(f"해시태그:{hashtag}에 대한 인기 게시물 수집 완료")
+
+
+
+
+
 @method_profiling
 def get_top_posts_success(hashtag):
 
@@ -60,7 +81,7 @@ def get_top_posts_success(hashtag):
             text_append = f"{data.name}\t{post.caption}\t{post.like_count}\t{post.comment_count}\t{post.hashtags}\t{post.shortcode}\t{post.url}\t{post.taken_at_timestamp}"
 
             if f.read():
-                ### 내용이 존재하면 받아온값을 추가해준다.
+                ### 내용이 존재하면 받아온값을 추가해준다.Q
                 f.write(f"\n{text_append}")
 
             else: