Integrasi Proxy dengan Google Cloud Functions: Pengaturan untuk Parsing dan Automatisasi

Google Cloud Functions adalah platform tanpa server untuk menjalankan kode tanpa perlu mengelola server. Saat bekerja dengan pengambilan data, otomatisasi permintaan API, atau pengumpulan data, sering kali diperlukan pengalihan lalu lintas melalui proxy untuk menghindari pemblokiran, rotasi IP, dan penargetan geografis. Dalam panduan ini, kita akan membahas pengaturan proxy di Cloud Functions menggunakan Python dan Node.js dengan contoh praktis.

Mengapa menggunakan proxy di Cloud Functions

Google Cloud Functions beroperasi dalam lingkungan terisolasi dengan alamat IP bersama dari pusat data Google. Saat melakukan permintaan sering ke API eksternal atau situs web, muncul beberapa masalah:

• Pemblokiran IP — banyak layanan (Google, Facebook, marketplace) mengenali lalu lintas dari pusat data dan menerapkan pembatasan kecepatan atau pemblokiran total.
• Pembatasan geografis — untuk mengakses konten yang hanya tersedia di negara tertentu (misalnya, pengambilan harga regional di Wildberries atau Ozon).
• Pembatasan kecepatan — satu alamat IP dapat melakukan jumlah permintaan terbatas per menit. Proxy memungkinkan distribusi beban.
• Anonymity — menyembunyikan sumber permintaan yang sebenarnya saat bekerja dengan data sensitif atau pengintaian kompetitif.

Skenario penggunaan proxy yang umum di Cloud Functions:

• Pengambilan data dari marketplace (Wildberries, Ozon, Amazon) untuk memantau harga pesaing
• Pengumpulan data dari media sosial (Instagram, TikTok) melalui API atau web scraping
• Otomatisasi pemeriksaan iklan di berbagai wilayah
• Permintaan massal ke mesin pencari (Google, Yandex) untuk analisis SEO
• Pengujian fungsi geolokasi aplikasi

Jenis proxy apa yang cocok untuk Cloud Functions

Pemilihan jenis proxy tergantung pada tugas, anggaran, dan kebutuhan akan anonimitas. Berikut adalah perbandingan opsi utama:

Rekomendasi pemilihan:

• Untuk pengambilan data dari marketplace (Wildberries, Ozon, Amazon) — proxy residensial dengan rotasi berdasarkan permintaan, agar setiap permintaan menggunakan IP baru.
• Untuk permintaan API (Google Maps API, OpenWeatherMap) — proxy data center dengan kecepatan tinggi, jika tidak ada batasan ketat pada IP.
• Untuk media sosial (Instagram, TikTok) — proxy mobile, karena mereka memiliki IP dari operator seluler dan jarang diblokir.
• Untuk pengambilan data SEO (Google, Yandex) — proxy residensial dengan penargetan geografis ke wilayah yang dibutuhkan.

Pengaturan proxy di Python (requests, aiohttp)

Python adalah bahasa yang paling populer untuk Cloud Functions saat bekerja dengan pengambilan data dan otomatisasi. Mari kita lihat integrasi proxy dengan pustaka requests (permintaan sinkron) dan aiohttp (permintaan asinkron).

Contoh dengan pustaka requests (HTTP-proxy)

import requests
import os

def parse_with_proxy(request):
    # Mendapatkan data proxy dari variabel lingkungan
    proxy_host = os.environ.get('PROXY_HOST', 'proxy.example.com')
    proxy_port = os.environ.get('PROXY_PORT', '8080')
    proxy_user = os.environ.get('PROXY_USER', 'username')
    proxy_pass = os.environ.get('PROXY_PASS', 'password')
    
    # Membentuk URL proxy dengan autentikasi
    proxy_url = f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}"
    
    proxies = {
        'http': proxy_url,
        'https': proxy_url
    }
    
    try:
        # Melakukan permintaan melalui proxy dengan timeout
        response = requests.get(
            'https://api.example.com/data',
            proxies=proxies,
            timeout=10,
            headers={'User-Agent': 'Mozilla/5.0'}
        )
        
        # Memeriksa status respons
        response.raise_for_status()
        
        return {
            'statusCode': 200,
            'body': response.json(),
            'ip_used': response.headers.get('X-Forwarded-For', 'unknown')
        }
        
    except requests.exceptions.ProxyError as e:
        return {'statusCode': 502, 'error': f'Kesalahan proxy: {str(e)}'}
    except requests.exceptions.Timeout:
        return {'statusCode': 504, 'error': 'Timeout permintaan'}
    except requests.exceptions.RequestException as e:
        return {'statusCode': 500, 'error': f'Permintaan gagal: {str(e)}'}

Poin penting:

• Variabel lingkungan — simpan data proxy (host, port, login, password) di Secret Manager atau variabel lingkungan Cloud Functions, bukan di kode.
• Timeout — pastikan untuk menetapkan timeout, agar fungsi tidak menggantung saat ada masalah dengan proxy.
• User-Agent — tambahkan header User-Agent, agar permintaan terlihat seperti berasal dari browser nyata.
• Penanganan kesalahan — tangani secara terpisah ProxyError (masalah dengan proxy) dan Timeout (proxy lambat).

Contoh dengan aiohttp (permintaan asinkron)

Untuk tugas yang sangat berat (misalnya, pengambilan 1000+ halaman), gunakan permintaan asinkron dengan aiohttp:

import aiohttp
import asyncio
import os

async def fetch_with_proxy(url, proxy_url):
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get(
                url,
                proxy=proxy_url,
                timeout=aiohttp.ClientTimeout(total=10),
                headers={'User-Agent': 'Mozilla/5.0'}
            ) as response:
                return await response.text()
        except aiohttp.ClientProxyConnectionError:
            return {'error': 'Koneksi proxy gagal'}
        except asyncio.TimeoutError:
            return {'error': 'Timeout permintaan'}

def parse_multiple_urls(request):
    proxy_url = f"http://{os.environ['PROXY_USER']}:{os.environ['PROXY_PASS']}@{os.environ['PROXY_HOST']}:{os.environ['PROXY_PORT']}"
    
    urls = [
        'https://example.com/page1',
        'https://example.com/page2',
        'https://example.com/page3'
    ]
    
    # Menjalankan permintaan asinkron secara paralel
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    tasks = [fetch_with_proxy(url, proxy_url) for url in urls]
    results = loop.run_until_complete(asyncio.gather(*tasks))
    
    return {'statusCode': 200, 'results': results}

Pendekatan asinkron memungkinkan melakukan 10-100 permintaan paralel melalui proxy, yang sangat penting untuk pengambilan data dalam jumlah besar dalam batas waktu eksekusi Cloud Functions (hingga 9 menit).

Bekerja dengan SOCKS5-proxy

Beberapa penyedia proxy menyediakan SOCKS5-proxy untuk bekerja lebih andal dengan lalu lintas UDP atau menghindari pemblokiran. Untuk bekerja dengan SOCKS5 di Python, gunakan pustaka requests[socks]:

# Tambahkan ke requirements.txt:
# requests[socks]

import requests

def use_socks5_proxy(request):
    proxy_url = f"socks5://{os.environ['PROXY_USER']}:{os.environ['PROXY_PASS']}@{os.environ['PROXY_HOST']}:{os.environ['PROXY_PORT']}"
    
    proxies = {
        'http': proxy_url,
        'https': proxy_url
    }
    
    response = requests.get(
        'https://api.ipify.org?format=json',
        proxies=proxies,
        timeout=10
    )
    
    return {'statusCode': 200, 'ip': response.json()}

Pengaturan proxy di Node.js (axios, node-fetch)

Node.js adalah bahasa kedua yang paling populer untuk Cloud Functions. Mari kita lihat integrasi proxy dengan pustaka axios dan node-fetch.

Contoh dengan axios

const axios = require('axios');
const HttpsProxyAgent = require('https-proxy-agent');

exports.parseWithProxy = async (req, res) => {
  const proxyUrl = `http://${process.env.PROXY_USER}:${process.env.PROXY_PASS}@${process.env.PROXY_HOST}:${process.env.PROXY_PORT}`;
  
  const agent = new HttpsProxyAgent(proxyUrl);
  
  try {
    const response = await axios.get('https://api.example.com/data', {
      httpsAgent: agent,
      timeout: 10000,
      headers: {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
      }
    });
    
    res.status(200).json({
      success: true,
      data: response.data,
      proxyUsed: proxyUrl.split('@')[1] // Mengembalikan host:port tanpa password
    });
    
  } catch (error) {
    if (error.code === 'ECONNREFUSED') {
      res.status(502).json({ error: 'Koneksi proxy ditolak' });
    } else if (error.code === 'ETIMEDOUT') {
      res.status(504).json({ error: 'Timeout proxy' });
    } else {
      res.status(500).json({ error: error.message });
    }
  }
};

Ketergantungan untuk package.json:

{
  "dependencies": {
    "axios": "^1.6.0",
    "https-proxy-agent": "^7.0.2"
  }
}

Contoh dengan node-fetch dan SOCKS5

const fetch = require('node-fetch');
const { SocksProxyAgent } = require('socks-proxy-agent');

exports.fetchWithSocks5 = async (req, res) => {
  const proxyUrl = `socks5://${process.env.PROXY_USER}:${process.env.PROXY_PASS}@${process.env.PROXY_HOST}:${process.env.PROXY_PORT}`;
  
  const agent = new SocksProxyAgent(proxyUrl);
  
  try {
    const response = await fetch('https://api.ipify.org?format=json', {
      agent,
      timeout: 10000
    });
    
    const data = await response.json();
    
    res.status(200).json({
      success: true,
      yourIP: data.ip
    });
    
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
};

Ketergantungan untuk SOCKS5:

{
  "dependencies": {
    "node-fetch": "^2.7.0",
    "socks-proxy-agent": "^8.0.2"
  }
}

Autentikasi proxy: login/password dan IP whitelist

Ada dua metode utama autentikasi saat bekerja dengan proxy:

1. Autentikasi dengan login dan password

Metode yang paling umum adalah mengirimkan kredensial dalam URL proxy:

http://username:password@proxy.example.com:8080

Keuntungan: Kemudahan pengaturan, tidak memerlukan IP sumber yang tetap.

Kekurangan: Kredensial dikirimkan dalam setiap permintaan, sedikit overhead.

2. Autentikasi dengan IP whitelist

Beberapa penyedia memungkinkan menambahkan alamat IP Cloud Functions ke whitelist. Masalahnya: Cloud Functions menggunakan IP dinamis dari pool Google Cloud.

Solusi: Gunakan Cloud NAT untuk mengarahkan lalu lintas keluar melalui IP eksternal statis:

1. Buat jaringan VPC dan subnet di Google Cloud
2. Atur Cloud NAT dengan reservasi IP statis
3. Hubungkan Cloud Functions ke VPC Connector
4. Tambahkan IP statis ke whitelist penyedia proxy

Setelah pengaturan, proxy tidak memerlukan login dan password:

proxies = {
    'http': 'http://proxy.example.com:8080',
    'https': 'http://proxy.example.com:8080'
}

Rekomendasi: Untuk sebagian besar kasus, gunakan autentikasi dengan login/password — ini lebih mudah dan tidak memerlukan biaya tambahan untuk Cloud NAT (mulai dari $0.044/jam + lalu lintas).

Rotasi IP dan manajemen pool proxy

Saat melakukan pengambilan data dalam jumlah besar, sangat penting untuk menggunakan rotasi IP agar terhindar dari pemblokiran. Ada beberapa pendekatan:

1. Rotasi di sisi penyedia (Rotating Proxies)

Banyak penyedia menawarkan rotating proxies — satu endpoint yang secara otomatis mengganti IP pada setiap permintaan atau berdasarkan timer:

# Satu endpoint, IP berubah secara otomatis
proxy_url = "http://username:password@rotating.proxy.com:8080"

# Setiap permintaan menggunakan IP baru
for i in range(100):
    response = requests.get('https://api.ipify.org', proxies={'http': proxy_url})
    print(f"Permintaan {i}: IP = {response.text}")

Keuntungan: Tidak perlu mengelola pool proxy secara manual, integrasi yang sederhana.

Kekurangan: Tidak ada kontrol atas IP tertentu, bisa lebih mahal.

2. Mengelola pool proxy secara manual

Jika Anda memiliki daftar proxy statis, implementasikan rotasi di tingkat kode:

import random
import requests

# Pool proxy (bisa dimuat dari Secret Manager)
PROXY_POOL = [
    "http://user:pass@proxy1.example.com:8080",
    "http://user:pass@proxy2.example.com:8080",
    "http://user:pass@proxy3.example.com:8080",
]

def get_random_proxy():
    return random.choice(PROXY_POOL)

def parse_with_rotation(urls):
    results = []
    
    for url in urls:
        proxy = get_random_proxy()
        
        try:
            response = requests.get(
                url,
                proxies={'http': proxy, 'https': proxy},
                timeout=10
            )
            results.append({
                'url': url,
                'status': response.status_code,
                'proxy': proxy.split('@')[1]
            })
        except Exception as e:
            # Jika proxy tidak berfungsi, coba proxy lain
            proxy = get_random_proxy()
            response = requests.get(url, proxies={'http': proxy, 'https': proxy})
            results.append({'url': url, 'status': response.status_code})
    
    return results

3. Proxy berbasis sesi (sticky sessions)

Untuk tugas yang memerlukan satu IP selama sesi (misalnya, otorisasi di situs), gunakan session ID dalam URL proxy:

# Menambahkan session ID ke login
import uuid

session_id = str(uuid.uuid4())
proxy_url = f"http://username-session-{session_id}:password@proxy.example.com:8080"

# Semua permintaan dengan session_id ini akan menggunakan satu IP
session = requests.Session()
session.proxies = {'http': proxy_url, 'https': proxy_url}

# Otorisasi
session.post('https://example.com/login', data={'user': 'test', 'pass': '123'})

# Permintaan berikutnya dalam sesi yang sama
session.get('https://example.com/dashboard')

Penanganan kesalahan dan timeout

Saat bekerja dengan proxy di Cloud Functions, sangat penting untuk menangani kesalahan dengan benar agar tidak kehilangan data dan tidak melebihi batas waktu eksekusi.

Jenis kesalahan dan cara penanganan

Contoh penanganan kesalahan yang komprehensif

import requests
import time
from requests.exceptions import ProxyError, Timeout, RequestException

def fetch_with_retry(url, proxy_pool, max_retries=3):
    """
    Permintaan dengan otomatis retry dan ganti proxy saat terjadi kesalahan
    """
    for attempt in range(max_retries):
        proxy = random.choice(proxy_pool)
        
        try:
            response = requests.get(
                url,
                proxies={'http': proxy, 'https': proxy},
                timeout=10,
                headers={'User-Agent': 'Mozilla/5.0'}
            )
            
            # Memeriksa status kode
            if response.status_code == 200:
                return {'success': True, 'data': response.text, 'proxy': proxy}
            elif response.status_code == 429:
                # Pembatasan kecepatan — tunggu dan coba lagi
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            elif response.status_code == 403:
                # IP diblokir — ganti proxy
                continue
            else:
                return {'success': False, 'status': response.status_code}
                
        except ProxyError:
            # Proxy tidak berfungsi — coba yang berikutnya
            print(f"Proxy {proxy} gagal, mencoba yang lain...")
            continue
        except Timeout:
            # Timeout — coba dengan proxy lain
            print(f"Timeout dengan {proxy}, mencoba lagi...")
            continue
        except RequestException as e:
            # Kesalahan lainnya
            print(f"Permintaan gagal: {e}")
            if attempt == max_retries - 1:
                return {'success': False, 'error': str(e)}
            continue
    
    return {'success': False, 'error': 'Maksimum percobaan terlampaui'}

Pengaturan timeout di Cloud Functions

Cloud Functions memiliki batas waktu eksekusi (secara default 60 detik, maksimum 540 detik). Pertimbangkan ini saat mengatur timeout proxy:

• Connection timeout — waktu untuk membangun koneksi dengan proxy (disarankan 5 detik)
• Read timeout — waktu untuk menerima respons dari server target melalui proxy (disarankan 10-15 detik)
• Total timeout — total waktu untuk seluruh permintaan (harus lebih kecil dari timeout fungsi)

# Python: timeout terpisah
response = requests.get(
    url,
    proxies=proxies,
    timeout=(5, 15)  # (connect timeout, read timeout)
)

# Node.js dengan axios
const response = await axios.get(url, {
  httpsAgent: agent,
  timeout: 10000  // total timeout dalam milidetik
});

Best practices dan optimasi kinerja

Rekomendasi untuk bekerja secara efektif dengan proxy di Cloud Functions:

1. Gunakan variabel lingkungan untuk kredensial

Jangan pernah menyimpan login dan password proxy dalam kode. Gunakan Secret Manager atau variabel lingkungan:

# Membuat rahasia di Google Cloud
gcloud secrets create proxy-credentials \
    --data-file=proxy-config.json

# Memberikan akses ke Cloud Functions
gcloud secrets add-iam-policy-binding proxy-credentials \
    --member=serviceAccount:PROJECT_ID@appspot.gserviceaccount.com \
    --role=roles/secretmanager.secretAccessor

# Membaca rahasia dalam kode
from google.cloud import secretmanager
import json

def get_proxy_config():
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{PROJECT_ID}/secrets/proxy-credentials/versions/latest"
    response = client.access_secret_version(request={"name": name})
    return json.loads(response.payload.data.decode('UTF-8'))

2. Cache hasil pengambilan data

Gunakan Cloud Storage atau Firestore untuk menyimpan data cache, agar tidak melakukan permintaan berulang melalui proxy:

import hashlib
from google.cloud import storage

def fetch_with_cache(url, proxy):
    # Menghasilkan kunci cache berdasarkan URL
    cache_key = hashlib.md5(url.encode()).hexdigest()
    
    # Memeriksa cache di Cloud Storage
    bucket = storage.Client().bucket('my-cache-bucket')
    blob = bucket.blob(f"cache/{cache_key}.json")
    
    if blob.exists():
        # Mengembalikan data yang dicache
        return json.loads(blob.download_as_text())
    
    # Melakukan permintaan melalui proxy
    response = requests.get(url, proxies={'http': proxy})
    data = response.json()
    
    # Menyimpan ke cache
    blob.upload_from_string(json.dumps(data))
    
    return data

3. Pemantauan dan logging

Pantau kinerja proxy dan frekuensi kesalahan melalui Cloud Logging:

import logging
import time

def fetch_with_logging(url, proxy):
    start_time = time.time()
    
    try:
        response = requests.get(url, proxies={'http': proxy}, timeout=10)
        duration = time.time() - start_time
        
        logging.info({
            'url': url,
            'proxy': proxy.split('@')[1],
            'status': response.status_code,
            'duration': duration,
            'success': True
        })
        
        return response
        
    except Exception as e:
        duration = time.time() - start_time
        
        logging.error({
            'url': url,
            'proxy': proxy.split('@')[1],
            'error': str(e),
            'duration': duration,
            'success': False
        })
        
        raise

4. Optimasi cold start

Cloud Functions memiliki latensi cold start. Minimalkan ketergantungan dan gunakan versi minimum pustaka:

# requirements.txt — hanya pustaka yang diperlukan
requests==2.31.0
# Hindari pustaka berat seperti pandas, jika tidak kritis

Gunakan variabel global untuk mendaur ulang koneksi:

# Membuat sesi satu kali saat cold start
session = requests.Session()
session.proxies = {'http': PROXY_URL, 'https': PROXY_URL}

def parse_data(request):
    # Mendaur ulang sesi antara panggilan
    response = session.get('https://api.example.com/data')
    return response.json()

5. Penargetan geografis proxy

Untuk tugas dengan penargetan geografis (misalnya, pengambilan harga regional), gunakan proxy yang terikat pada negara atau kota tertentu:

# Contoh dengan proxy residensial, di mana negara dapat ditentukan dalam login
proxy_url = f"http://username-country-ru:password@proxy.example.com:8080"

# Atau menggunakan endpoint berbeda untuk negara yang berbeda
PROXIES_BY_COUNTRY = {
    'RU': 'http://user:pass@ru.proxy.example.com:8080',
    'US': 'http://user:pass@us.proxy.example.com:8080',
    'DE': 'http://user:pass@de.proxy.example.com:8080'
}

def parse_by_country(country_code):
    proxy = PROXIES_BY_COUNTRY.get(country_code)
    response = requests.get('https://example.com', proxies={'http': proxy})
    return response.text

Kesimpulan

Integrasi proxy dengan Google Cloud Functions membuka banyak peluang untuk pengambilan data, otomatisasi, dan bekerja dengan API tanpa batasan IP. Poin utama yang perlu diperhatikan: penanganan kesalahan yang benar dengan logika retry, penggunaan timeout untuk mencegah penggantung, rotasi IP untuk menghindari pemblokiran, dan penyimpanan kredensial dengan aman di Secret Manager.

Untuk sebagian besar tugas pengambilan data dan otomatisasi, pilihan terbaik adalah proxy residensial — mereka memberikan anonimitas tinggi dan persentase pemblokiran rendah berkat penggunaan IP pengguna nyata. Untuk bekerja dengan media sosial dan aplikasi mobile, kami merekomendasikan proxy mobile, yang memiliki IP dari operator seluler dan hampir tidak diblokir oleh platform seperti Instagram dan TikTok.

Dengan pengaturan yang tepat, Cloud Functions dengan proxy memberikan solusi yang dapat diskalakan dan hemat biaya untuk memproses data dalam jumlah besar tanpa perlu mengelola infrastruktur.