1
2
3

Computer Science and Engineering
└─ DevOps and Platform Engineering
   └─ CI/CD, DevOps, Monitoring and Observability, Version Control Systems, etc

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

import os
import subprocess

def run_tests():
    print('테스트 실행 중...')
    result = subprocess.run(['pytest'], capture_output=True)
    return result.returncode == 0

def deploy():
    print('배포 시작...')
    os.system('kubectl apply -f deployment.yaml')
    print('배포 완료')

if __name__ == '__main__':
    if run_tests():
        deploy()
    else:
        print('테스트 실패, 배포 중단')

1
2

Computer Science and Engineering
└─ DevOps and Platform Engineering

 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

name: CI/CD

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  build-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: 3.10
      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
      - name: Run tests
        run: pytest
      - name: Code analysis
        run: bandit -r src/
      - name: Build Docker image
        run: docker build -t myapp:${{ github.sha }} .
      - name: Push to registry
        run: docker push myapp:${{ github.sha }}
      - name: Deploy to K8s
        env:
          IMAGE: myapp:${{ github.sha }}
        run: |
          kubectl set image deployment/myapp myapp=$IMAGE
          kubectl rollout status deployment/myapp

 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

apiVersion: templates.gatekeeper.sh/v1beta1
kind: ConstraintTemplate
metadata:
  name: k8srequiredlabel
spec:
  crd:
    spec:
      names:
        kind: K8sRequiredLabel
  targets:
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package k8srequiredlabel
        violation[{"msg": msg}] {
          input.review.object.metadata.labels["team"] == ""
          msg := "missing label team"
        }
---
apiVersion: constraints.gatekeeper.sh/v1beta1
kind: K8sRequiredLabel
metadata:
  name: require-team-label
spec:
  match:
    kinds:
      - apiGroups: [""]
        kinds: ["Namespace"]

 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

apiVersion: kyverno.io/v1
kind: ClusterPolicy
metadata:
  name: enforce-pod-label
spec:
  rules:
    - name: check-pod-label
      match:
        resources:
          kinds: ["Pod"]
      validate:
        message: "Pods must have 'app' label"
        pattern:
          metadata:
            labels:
              app: "?*"
    - name: add-image-digest
      match:
        resources:
          kinds: ["Pod"]
      mutate:
        patchStrategicMerge:
          spec:
            containers:
              - (name): "*"
                image: "{{@}}@sha256:{{ digest (image) }}"

1
2

개발자 코드 커밋 → Jenkins 빌드 트리거 → 단위 테스트 실행 → 
Docker 이미지 생성 → Kubernetes 클러스터 배포 → 상태 모니터링

1
2

Terraform 스크립트 작성 → AWS 리소스 정의 → 
terraform apply 실행 → 인프라 자동 생성 → 상태 파일 관리

1
2

Docker 이미지 빌드 → 컨테이너 레지스트리 푸시 → 
Kubernetes 배포 → 오토 스케일링 → 롤링 업데이트

  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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486

#!/usr/bin/env python3
"""
DevOps Pipeline Implementation Example
Netflix-inspired CI/CD Pipeline with monitoring and automated deployment
"""

import json
import time
import subprocess
import logging
from datetime import datetime
from dataclasses import dataclass
from typing import List, Dict, Any
from enum import Enum

# 로깅 설정
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class DeploymentStrategy(Enum):
    """배포 전략 열거형"""
    BLUE_GREEN = "blue_green"
    CANARY = "canary"
    ROLLING = "rolling"

class PipelineStage(Enum):
    """파이프라인 단계 열거형"""
    SOURCE = "source"
    BUILD = "build"
    TEST = "test"
    SECURITY_SCAN = "security_scan"
    DEPLOY_STAGING = "deploy_staging"
    INTEGRATION_TEST = "integration_test"
    DEPLOY_PRODUCTION = "deploy_production"
    MONITOR = "monitor"

@dataclass
class BuildArtifact:
    """빌드 아티팩트 정보"""
    name: str
    version: str
    image_url: str
    created_at: datetime
    test_results: Dict[str, Any]

@dataclass
class DeploymentConfig:
    """배포 설정"""
    strategy: DeploymentStrategy
    target_environment: str
    replicas: int
    health_check_url: str
    rollback_threshold: float = 0.95

class HealthChecker:
    """서비스 헬스 체크 기능"""
    
    @staticmethod
    def check_service_health(url: str, expected_status: int = 200) -> bool:
        """서비스 상태 확인"""
        try:
            # 실제 구현에서는 requests 라이브러리 사용
            logger.info(f"Health check for {url}")
            # 시뮬레이션: 90% 확률로 성공
            import random
            return random.random() > 0.1
        except Exception as e:
            logger.error(f"Health check failed: {e}")
            return False

class MetricsCollector:
    """메트릭 수집 및 모니터링"""
    
    def __init__(self):
        self.metrics = {}
    
    def collect_deployment_metrics(self, deployment_id: str) -> Dict[str, float]:
        """배포 메트릭 수집"""
        # 시뮬레이션 데이터
        import random
        metrics = {
            "cpu_usage": random.uniform(20, 80),
            "memory_usage": random.uniform(30, 70),
            "response_time": random.uniform(100, 500),
            "error_rate": random.uniform(0, 5),
            "throughput": random.uniform(1000, 5000)
        }
        self.metrics[deployment_id] = metrics
        return metrics
    
    def check_sla_compliance(self, metrics: Dict[str, float]) -> bool:
        """SLA 준수 여부 확인"""
        sla_thresholds = {
            "cpu_usage": 85,
            "memory_usage": 80,
            "response_time": 1000,
            "error_rate": 1
        }
        
        for metric, value in metrics.items():
            if metric in sla_thresholds and value > sla_thresholds[metric]:
                logger.warning(f"SLA violation: {metric} = {value}")
                return False
        return True

class ChaosEngineering:
    """카오스 엔지니어링 - Netflix Chaos Monkey 스타일"""
    
    @staticmethod
    def simulate_instance_failure(instance_id: str) -> bool:
        """인스턴스 장애 시뮬레이션"""
        logger.info(f"Simulating failure for instance {instance_id}")
        # 실제로는 특정 인스턴스를 종료하거나 네트워크를 차단
        time.sleep(2)  # 장애 시뮬레이션
        return True
    
    @staticmethod
    def test_circuit_breaker(service_name: str) -> bool:
        """서킷 브레이커 테스트"""
        logger.info(f"Testing circuit breaker for {service_name}")
        # 서킷 브레이커 동작 확인
        return True

class DevOpsPipeline:
    """Netflix 스타일 DevOps 파이프라인"""
    
    def __init__(self, application_name: str):
        self.application_name = application_name
        self.health_checker = HealthChecker()
        self.metrics_collector = MetricsCollector()
        self.chaos_engineering = ChaosEngineering()
        self.pipeline_history = []
    
    def run_pipeline(self, source_code_url: str, deployment_config: DeploymentConfig) -> bool:
        """전체 파이프라인 실행"""
        pipeline_id = f"{self.application_name}-{int(time.time())}"
        logger.info(f"Starting pipeline {pipeline_id}")
        
        try:
            # 1. Source Stage
            if not self._source_stage(source_code_url):
                raise Exception("Source stage failed")
            
            # 2. Build Stage
            artifact = self._build_stage()
            if not artifact:
                raise Exception("Build stage failed")
            
            # 3. Test Stage
            if not self._test_stage(artifact):
                raise Exception("Test stage failed")
            
            # 4. Security Scan Stage
            if not self._security_scan_stage(artifact):
                raise Exception("Security scan failed")
            
            # 5. Deploy to Staging
            staging_deployment = self._deploy_stage(artifact, "staging", deployment_config)
            if not staging_deployment:
                raise Exception("Staging deployment failed")
            
            # 6. Integration Tests
            if not self._integration_test_stage("staging"):
                raise Exception("Integration tests failed")
            
            # 7. Chaos Engineering Tests
            if not self._chaos_testing_stage("staging"):
                logger.warning("Chaos tests failed, but continuing...")
            
            # 8. Production Deployment
            prod_deployment = self._deploy_stage(artifact, "production", deployment_config)
            if not prod_deployment:
                raise Exception("Production deployment failed")
            
            # 9. Post-deployment Monitoring
            if not self._monitoring_stage(prod_deployment):
                logger.warning("Monitoring alerts detected")
                return self._rollback_deployment(prod_deployment)
            
            logger.info(f"Pipeline {pipeline_id} completed successfully")
            self.pipeline_history.append({
                "pipeline_id": pipeline_id,
                "status": "success",
                "timestamp": datetime.now(),
                "artifact": artifact
            })
            return True
            
        except Exception as e:
            logger.error(f"Pipeline {pipeline_id} failed: {e}")
            self.pipeline_history.append({
                "pipeline_id": pipeline_id,
                "status": "failed",
                "timestamp": datetime.now(),
                "error": str(e)
            })
            return False
    
    def _source_stage(self, source_url: str) -> bool:
        """소스 코드 가져오기"""
        logger.info(f"Fetching source code from {source_url}")
        # Git clone 시뮬레이션
        time.sleep(1)
        return True
    
    def _build_stage(self) -> BuildArtifact:
        """빌드 단계 - Docker 이미지 생성"""
        logger.info("Building application...")
        
        # 빌드 프로세스 시뮬레이션
        time.sleep(3)
        
        # 버전 생성 (Semantic Versioning)
        version = f"1.{int(time.time() % 1000)}.0"
        
        artifact = BuildArtifact(
            name=self.application_name,
            version=version,
            image_url=f"registry.company.com/{self.application_name}:{version}",
            created_at=datetime.now(),
            test_results={}
        )
        
        logger.info(f"Build completed: {artifact.image_url}")
        return artifact
    
    def _test_stage(self, artifact: BuildArtifact) -> bool:
        """테스트 단계 - 단위 테스트, 통합 테스트"""
        logger.info("Running automated tests...")
        
        # 다양한 테스트 실행 시뮬레이션
        test_results = {
            "unit_tests": {"passed": 95, "failed": 2, "coverage": 87.5},
            "integration_tests": {"passed": 23, "failed": 0},
            "performance_tests": {"avg_response_time": 245, "max_throughput": 4500}
        }
        
        artifact.test_results = test_results
        
        # 테스트 실패 기준 확인
        if test_results["unit_tests"]["failed"] > 5:
            logger.error("Too many unit test failures")
            return False
        
        if test_results["unit_tests"]["coverage"] < 80:
            logger.error("Code coverage below threshold")
            return False
        
        logger.info("All tests passed")
        return True
    
    def _security_scan_stage(self, artifact: BuildArtifact) -> bool:
        """보안 스캔 단계 - DevSecOps"""
        logger.info("Running security scans...")
        
        # 보안 스캔 시뮬레이션
        time.sleep(2)
        
        security_results = {
            "vulnerability_scan": {"critical": 0, "high": 1, "medium": 3, "low": 5},
            "dependency_check": {"vulnerable_dependencies": 2},
            "static_analysis": {"security_hotspots": 1}
        }
        
        # 보안 정책 확인
        if security_results["vulnerability_scan"]["critical"] > 0:
            logger.error("Critical vulnerabilities found")
            return False
        
        if security_results["vulnerability_scan"]["high"] > 3:
            logger.error("Too many high-severity vulnerabilities")
            return False
        
        logger.info("Security scan passed")
        return True
    
    def _deploy_stage(self, artifact: BuildArtifact, environment: str, 
                     config: DeploymentConfig) -> str:
        """배포 단계"""
        deployment_id = f"{artifact.name}-{environment}-{int(time.time())}"
        logger.info(f"Deploying to {environment} with strategy {config.strategy.value}")
        
        if config.strategy == DeploymentStrategy.CANARY:
            return self._canary_deployment(artifact, environment, deployment_id, config)
        elif config.strategy == DeploymentStrategy.BLUE_GREEN:
            return self._blue_green_deployment(artifact, environment, deployment_id, config)
        else:
            return self._rolling_deployment(artifact, environment, deployment_id, config)
    
    def _canary_deployment(self, artifact: BuildArtifact, environment: str, 
                          deployment_id: str, config: DeploymentConfig) -> str:
        """카나리 배포"""
        logger.info("Starting canary deployment...")
        
        # 1단계: 5% 트래픽으로 시작
        logger.info("Deploying to 5% of instances...")
        time.sleep(2)
        
        # 초기 메트릭 확인
        metrics = self.metrics_collector.collect_deployment_metrics(deployment_id)
        if not self.metrics_collector.check_sla_compliance(metrics):
            logger.error("Canary deployment failed SLA check")
            return None
        
        # 2단계: 25% 트래픽
        logger.info("Scaling to 25% of traffic...")
        time.sleep(2)
        
        # 3단계: 100% 트래픽
        logger.info("Completing canary deployment to 100%")
        time.sleep(2)
        
        logger.info(f"Canary deployment {deployment_id} completed")
        return deployment_id
    
    def _blue_green_deployment(self, artifact: BuildArtifact, environment: str,
                              deployment_id: str, config: DeploymentConfig) -> str:
        """블루-그린 배포"""
        logger.info("Starting blue-green deployment...")
        
        # Green 환경에 배포
        logger.info("Deploying to green environment...")
        time.sleep(3)
        
        # Green 환경 검증
        if not self.health_checker.check_service_health(config.health_check_url):
            logger.error("Green environment health check failed")
            return None
        
        # 트래픽 전환
        logger.info("Switching traffic from blue to green...")
        time.sleep(1)
        
        logger.info(f"Blue-green deployment {deployment_id} completed")
        return deployment_id
    
    def _rolling_deployment(self, artifact: BuildArtifact, environment: str,
                           deployment_id: str, config: DeploymentConfig) -> str:
        """롤링 배포"""
        logger.info("Starting rolling deployment...")
        
        # 인스턴스별 순차 배포
        for i in range(config.replicas):
            logger.info(f"Updating instance {i+1}/{config.replicas}")
            time.sleep(1)
            
            # 각 인스턴스 업데이트 후 상태 확인
            if not self.health_checker.check_service_health(config.health_check_url):
                logger.error(f"Instance {i+1} health check failed")
                return None
        
        logger.info(f"Rolling deployment {deployment_id} completed")
        return deployment_id
    
    def _integration_test_stage(self, environment: str) -> bool:
        """통합 테스트 단계"""
        logger.info(f"Running integration tests in {environment}")
        
        # API 테스트, E2E 테스트 시뮬레이션
        test_scenarios = [
            "user_registration_flow",
            "payment_processing",
            "content_streaming",
            "recommendation_engine"
        ]
        
        for scenario in test_scenarios:
            logger.info(f"Testing scenario: {scenario}")
            time.sleep(0.5)
            # 시뮬레이션: 95% 성공률
            import random
            if random.random() < 0.05:
                logger.error(f"Integration test failed: {scenario}")
                return False
        
        logger.info("All integration tests passed")
        return True
    
    def _chaos_testing_stage(self, environment: str) -> bool:
        """카오스 엔지니어링 테스트"""
        logger.info(f"Running chaos tests in {environment}")
        
        # 인스턴스 장애 테스트
        self.chaos_engineering.simulate_instance_failure("instance-1")
        
        # 서킷 브레이커 테스트
        self.chaos_engineering.test_circuit_breaker("recommendation-service")
        
        # 시스템 복구 확인
        time.sleep(3)
        if not self.health_checker.check_service_health("http://staging.company.com/health"):
            logger.error("System did not recover from chaos test")
            return False
        
        logger.info("Chaos tests completed successfully")
        return True
    
    def _monitoring_stage(self, deployment_id: str) -> bool:
        """배포 후 모니터링"""
        logger.info(f"Monitoring deployment {deployment_id}")
        
        # 모니터링 기간 (실제로는 더 길 수 있음)
        for i in range(3):
            logger.info(f"Monitoring check {i+1}/3")
            time.sleep(2)
            
            metrics = self.metrics_collector.collect_deployment_metrics(deployment_id)
            if not self.metrics_collector.check_sla_compliance(metrics):
                logger.error("SLA violation detected during monitoring")
                return False
        
        logger.info("Monitoring completed - all metrics within SLA")
        return True
    
    def _rollback_deployment(self, deployment_id: str) -> bool:
        """배포 롤백"""
        logger.warning(f"Rolling back deployment {deployment_id}")
        
        # 이전 버전으로 롤백
        time.sleep(3)
        
        # 롤백 후 상태 확인
        if self.health_checker.check_service_health("http://production.company.com/health"):
            logger.info("Rollback completed successfully")
            return True
        else:
            logger.error("Rollback failed")
            return False
    
    def get_pipeline_status(self) -> Dict[str, Any]:
        """파이프라인 상태 조회"""
        return {
            "application": self.application_name,
            "total_runs": len(self.pipeline_history),
            "success_rate": len([h for h in self.pipeline_history if h["status"] == "success"]) / max(len(self.pipeline_history), 1),
            "recent_runs": self.pipeline_history[-5:] if self.pipeline_history else []
        }

# 사용 예시
def main():
    """DevOps 파이프라인 실행 예시"""
    
    # Netflix 스타일 마이크로서비스 배포
    pipeline = DevOpsPipeline("user-service")
    
    # 카나리 배포 설정
    canary_config = DeploymentConfig(
        strategy=DeploymentStrategy.CANARY,
        target_environment="production",
        replicas=10,
        health_check_url="http://user-service.company.com/health",
        rollback_threshold=0.95
    )
    
    # 파이프라인 실행
    logger.info("=== Starting DevOps Pipeline ===")
    success = pipeline.run_pipeline(
        source_code_url="https://github.com/company/user-service.git",
        deployment_config=canary_config
    )
    
    if success:
        logger.info("🎉 Pipeline completed successfully!")
    else:
        logger.error("❌ Pipeline failed!")
    
    # 파이프라인 상태 출력
    status = pipeline.get_pipeline_status()
    logger.info(f"Pipeline Status: {json.dumps(status, indent=2, default=str)}")
    
    # 블루-그린 배포 예시
    logger.info("\n=== Running Blue-Green Deployment ===")
    blue_green_config = DeploymentConfig(
        strategy=DeploymentStrategy.BLUE_GREEN,
        target_environment="production",
        replicas=5,
        health_check_url="http://user-service.company.com/health"
    )
    
    pipeline.run_pipeline(
        source_code_url="https://github.com/company/user-service.git",
        deployment_config=blue_green_config
    )

if __name__ == "__main__":
    main()

1
2

[개발자] ──(코드 커밋)──> [버전 관리] ──> [CI/CD 파이프라인] ──> [테스트] ──> [배포] ──> [운영 환경]
         └─(피드백)───────────────────────────────────────────────────────────────┘

1
2
3

[개발자] ──> [버전 관리(Git)] ──> [CI/CD 파이프라인] ──> [테스트] ──> [배포] ──> [운영 환경]
         └─(피드백)───────────────────────────────────────────────────────────────┘
[모니터링/로깅] ──> [피드백] ──> [개발자]

1
2
3

[개발자] ──> [GitHub] ──> [Jenkins] ──> [Docker] ──> [Kubernetes] ──> [AWS]
         └─(피드백)───────────────────────────────────────────────┘
[Prometheus] ──> [피드백] ──> [개발자]

1
2
3
4
5
6
7

# Jenkins API를 활용한 파이프라인 트리거 예시 (간단한 HTTP 요청)
import requests

jenkins_url = "http://your-jenkins-server/job/your-job/build"
auth = ('username', 'api-token')
response = requests.post(jenkins_url, auth=auth)
print("Pipeline triggered:", response.status_code)